Odoo-Modules/docs/superpowers/plans/2026-05-27-nexacloud-billing-importer.md

NexaCloud → Odoo Billing Importer (Sub-project #2a) — Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Build a one-time, re-runnable, read-only importer that backfills NexaCloud customers/plans/deployments into Odoo as a shadow copy (drafts, no charge) for dual-run reconciliation.

Architecture: A fusion.billing.import.wizard transient model. _read_nexacloud_rows() opens a read-only psycopg2 connection (DSN from ir.config_parameter) and returns plain row dicts — the only code touching NexaCloud. _import_rows(data, dry_run) is pure Odoo: it upserts the nexacloud service, a cpu_seconds metric, Monthly/Yearly recurrences, partners+links (reusing _resolve_or_create_partner), a per-plan catalog (product + CPU-overage product + fusion.billing.charge with plan_id left NULL), and one draft shadow sale.order per deployment with the flat price set explicitly on the line. Shadow-safety holds by construction: draft + no payment token + charge plan_id NULL.

Tech Stack: Odoo 19 Enterprise (Python 3.12), sale_subscription, account_accountant, payment_stripe, psycopg2. Tests: odoo.tests.common.TransactionCase on odoo-trial.

Spec: docs/superpowers/specs/2026-05-27-nexacloud-billing-importer-design.md

---

Conventions for every task

• Never code Odoo internals from memory (repo CLAUDE.md rule #1). The uncertain internals (recurring_invoice, is_subscription on a draft order, sale.subscription.plan fields, price_unit stickiness, sale.subscription.plan billing_period_unit values) are verified by the tests themselves on odoo-trial — when a test fails because an assumption is wrong, fix the source, do not weaken the assertion.
• Models, not UI: all logic lives in _import_rows / _do_import / _import_* model methods; the wizard button only calls them. This keeps everything testable under TransactionCase.
• Money: CAD, prices are Float/Monetary. CPU overage: price_per_unit=0.0075, unit_batch=3600.
• New fields on native models: x_fc_* prefix.
• Registering tests: append from . import test_importer to tests/__init__.py in the task that creates it; commit __init__.py alongside so the package always imports.

Test environment

Tests run on odoo-trial (Proxmox VM 316, Odoo 19 Enterprise, db trial) — local dev is Community and cannot install this module. One runner:

bash scripts/fcb_test_on_trial.sh

• It re-syncs the module to the sandbox and runs -u fusion_centralize_billing --test-enable --test-tags /fusion_centralize_billing.
• Pass condition: output contains FCB_EXIT=0.
• The script runs the whole FCB suite (it cannot target one test); every "run the test" step below means "run the suite, ~1–2 min".
• Never run --test-enable against production nexamain.

File structure (this plan)

fusion_centralize_billing/
  __init__.py                 # + from . import wizards
  models/
    __init__.py               # + from . import res_partner
    sale_order.py             # + x_fc_* fields on the existing SaleOrder inherit
    res_partner.py            # NEW: x_fc_stripe_customer_id
  wizards/
    __init__.py               # NEW
    import_wizard.py          # NEW: the importer (read + import logic)
  views/
    import_wizard_views.xml   # NEW: wizard form + action + menu
  security/
    ir.model.access.csv       # + wizard ACL line
  __manifest__.py             # + views file
  tests/
    __init__.py               # + from . import test_importer
    test_importer.py          # NEW

---

Task 1: Scaffolding — x_fc fields, partner inherit, wizard skeleton, security, manifest

Files:

• Modify: fusion_centralize_billing/models/sale_order.py

• Create: fusion_centralize_billing/models/res_partner.py

• Modify: fusion_centralize_billing/models/__init__.py

• Create: fusion_centralize_billing/wizards/__init__.py

• Create: fusion_centralize_billing/wizards/import_wizard.py

• Create: fusion_centralize_billing/views/import_wizard_views.xml

• Modify: fusion_centralize_billing/__init__.py

• Modify: fusion_centralize_billing/security/ir.model.access.csv

• Modify: fusion_centralize_billing/__manifest__.py

• Step 1: Add x_fc_* fields to the existing sale.order inherit

In models/sale_order.py, add these fields to the SaleOrder class (keep _fc_rate_usage):

    x_fc_nexacloud_subscription_id = fields.Char(
        index=True, copy=False,
        help="Source NexaCloud subscription id — the importer's idempotency key.")
    x_fc_nexacloud_deployment_id = fields.Char(index=True, copy=False)
    x_fc_billing_service_id = fields.Many2one(
        "fusion.billing.service", index=True, copy=False, ondelete="set null")
    x_fc_shadow = fields.Boolean(
        default=False, copy=False,
        help="Imported in shadow mode: Odoo computes but must not charge/post/email.")

• Step 2: Create the res.partner inherit

fusion_centralize_billing/models/res_partner.py:

# -*- coding: utf-8 -*-
# Copyright 2026 Nexa Systems Inc.
# License OPL-1
from odoo import fields, models


class ResPartner(models.Model):
    _inherit = "res.partner"

    x_fc_stripe_customer_id = fields.Char(
        index=True, copy=False,
        help="Existing Stripe customer id imported from a source app, reused at flip.")

Append to models/__init__.py: from . import res_partner.

• Step 3: Create the wizard skeleton

fusion_centralize_billing/wizards/__init__.py:

from . import import_wizard

fusion_centralize_billing/wizards/import_wizard.py:

# -*- coding: utf-8 -*-
# Copyright 2026 Nexa Systems Inc.
# License OPL-1
import json
import logging

from odoo import api, fields, models
from odoo.exceptions import UserError

_logger = logging.getLogger(__name__)

NEXACLOUD_CODE = "nexacloud"
CPU_METRIC_CODE = "cpu_seconds"
CPU_RATE_PER_CORE_HOUR = 0.0075      # NexaCloud CPU rate, CAD per core-hour
CPU_SECONDS_PER_CORE_HOUR = 3600.0   # one core-hour = 3600 cpu-seconds


class FusionBillingImportWizard(models.TransientModel):
    _name = "fusion.billing.import.wizard"
    _description = "Fusion Billing — NexaCloud Importer"

    dry_run = fields.Boolean(
        default=True,
        help="Read and report what would be imported, without writing anything.")
    result_summary = fields.Text(readonly=True)

    def action_run_import(self):
        self.ensure_one()
        data = self._read_nexacloud_rows()
        summary = self._import_rows(data, dry_run=self.dry_run)
        self.result_summary = json.dumps(summary, indent=2, default=str)
        return {
            "type": "ir.actions.act_window",
            "res_model": self._name,
            "res_id": self.id,
            "view_mode": "form",
            "target": "new",
        }

    # ----- read side (the ONLY code that touches NexaCloud) ------------------
    def _read_nexacloud_rows(self):
        """Open a READ-ONLY psycopg2 connection to the nexacloud Postgres (DSN in
        ir.config_parameter 'fusion_billing.nexacloud_dsn') and return rows as dicts.
        Raises UserError on a missing DSN or a failed connection."""
        import psycopg2
        import psycopg2.extras
        dsn = self.env["ir.config_parameter"].sudo().get_param("fusion_billing.nexacloud_dsn")
        if not dsn:
            raise UserError(
                "NexaCloud DSN not configured. Set the 'fusion_billing.nexacloud_dsn' "
                "system parameter to a read-only Postgres connection string.")
        try:
            conn = psycopg2.connect(dsn)
        except Exception as e:  # noqa: BLE001 - surface as a user error
            raise UserError("Could not connect to the NexaCloud database: %s" % e)
        try:
            conn.set_session(readonly=True)
            cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
            data = {}
            cur.execute(
                "SELECT id, email, full_name, company, billing_email, billing_address, "
                "billing_city, billing_state, billing_postal_code, billing_country, "
                "tax_id, stripe_customer_id FROM users")
            data["users"] = [dict(r) for r in cur.fetchall()]
            cur.execute(
                "SELECT id, name, price_monthly, price_yearly, cpu_seconds_quota, "
                "is_active FROM plans")
            data["plans"] = [dict(r) for r in cur.fetchall()]
            cur.execute(
                "SELECT id, user_id, deployment_id, plan_id, status, billing_cycle, "
                "current_period_start, current_period_end FROM subscriptions")
            data["subscriptions"] = [dict(r) for r in cur.fetchall()]
            return data
        finally:
            conn.close()

    # ----- import side (pure Odoo; unit-tested) ------------------------------
    @api.model
    def _import_rows(self, data, dry_run=False):
        """Upsert NexaCloud rows into Odoo. Idempotent. With dry_run=True the writes
        happen inside a savepoint that is rolled back, so nothing persists."""
        if not dry_run:
            return self._do_import(data)
        result = {}

        class _Rollback(Exception):
            pass

        try:
            with self.env.cr.savepoint():
                result.update(self._do_import(data))
                raise _Rollback()
        except _Rollback:
            pass
        result["dry_run"] = True
        return result

    @api.model
    def _do_import(self, data):
        return {"created": {}, "updated": {}, "skipped": [], "failed": []}

• Step 4: Add the wizard view + action + menu

fusion_centralize_billing/views/import_wizard_views.xml:

<?xml version="1.0" encoding="utf-8"?>
<odoo>
    <record id="view_fusion_billing_import_wizard_form" model="ir.ui.view">
        <field name="name">fusion.billing.import.wizard.form</field>
        <field name="model">fusion.billing.import.wizard</field>
        <field name="arch" type="xml">
            <form string="Import from NexaCloud">
                <group>
                    <field name="dry_run"/>
                </group>
                <group string="Result" invisible="not result_summary">
                    <field name="result_summary" nolabel="1" widget="text"/>
                </group>
                <footer>
                    <button name="action_run_import" type="object" string="Run Import"
                            class="btn-primary"/>
                    <button string="Close" class="btn-secondary" special="cancel"/>
                </footer>
            </form>
        </field>
    </record>

    <record id="action_fusion_billing_import_wizard" model="ir.actions.act_window">
        <field name="name">Import from NexaCloud</field>
        <field name="res_model">fusion.billing.import.wizard</field>
        <field name="view_mode">form</field>
        <field name="target">new</field>
    </record>

    <menuitem id="menu_fusion_billing_root" name="Fusion Billing"
              parent="account.menu_finance" sequence="90"/>
    <menuitem id="menu_fusion_billing_import" name="Import from NexaCloud"
              parent="menu_fusion_billing_root"
              action="action_fusion_billing_import_wizard" sequence="10"
              groups="base.group_system"/>
</odoo>

• Step 5: Wire module imports, security, manifest

Append to fusion_centralize_billing/__init__.py: from . import wizards. (Confirm it already has from . import models and from . import controllers; add the wizards line.)

Append to security/ir.model.access.csv:

access_fusion_billing_import_wizard,fusion.billing.import.wizard,model_fusion_billing_import_wizard,base.group_system,1,1,1,1

In __manifest__.py, add the view to data (after the cron):

    "data": [
        "security/ir.model.access.csv",
        "data/ir_cron.xml",
        "views/import_wizard_views.xml",
    ],

• Step 6: Verify the module upgrades cleanly on odoo-trial

Run: bash scripts/fcb_test_on_trial.sh Expected: FCB_EXIT=0 (the 39 existing tests still pass; new model/fields/view load with no traceback).

• Step 7: Commit

git add fusion_centralize_billing/models/sale_order.py fusion_centralize_billing/models/res_partner.py fusion_centralize_billing/models/__init__.py fusion_centralize_billing/wizards/ fusion_centralize_billing/views/import_wizard_views.xml fusion_centralize_billing/__init__.py fusion_centralize_billing/security/ir.model.access.csv fusion_centralize_billing/__manifest__.py
git commit -m "feat(billing): importer scaffold — x_fc fields, wizard, security, view"

---

Task 2: Identity import (users → partners + links)

Files:

• Modify: fusion_centralize_billing/wizards/import_wizard.py

• Create: fusion_centralize_billing/tests/test_importer.py

• Modify: fusion_centralize_billing/tests/__init__.py

• Step 1: Register + write the failing test

Append to tests/__init__.py: from . import test_importer.

fusion_centralize_billing/tests/test_importer.py:

# -*- coding: utf-8 -*-
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

• Step 2: Run it, expect failure

Run: bash scripts/fcb_test_on_trial.sh Expected: FAIL — _do_import returns the empty stub; no partners/links created.

• Step 3: Implement service/metric/recurrence helpers + user import

Replace the stub _do_import and add helpers in wizards/import_wizard.py:

    @api.model
    def _fc_service(self):
        Service = self.env['fusion.billing.service']
        svc = Service.search([('code', '=', NEXACLOUD_CODE)], limit=1)
        return svc or Service.create({'name': 'NexaCloud', 'code': NEXACLOUD_CODE})

    @api.model
    def _fc_cpu_metric(self):
        Metric = self.env['fusion.billing.metric']
        m = Metric.search([('code', '=', CPU_METRIC_CODE)], limit=1)
        return m or Metric.create({
            'name': 'CPU seconds', 'code': CPU_METRIC_CODE,
            'aggregation': 'sum', 'unit_label': 'CPU-seconds'})

    @api.model
    def _fc_recurrence_plan(self, unit):
        Plan = self.env['sale.subscription.plan']
        plan = Plan.search([('billing_period_value', '=', 1),
                            ('billing_period_unit', '=', unit)], limit=1)
        if plan:
            return plan
        label = 'Monthly' if unit == 'month' else 'Yearly'
        return Plan.create({'name': label, 'billing_period_value': 1,
                            'billing_period_unit': unit})

    @api.model
    def _fc_resolve_country(self, value):
        Country = self.env['res.country']
        if not value:
            return Country.browse()
        v = value.strip()
        return Country.search(['|', ('code', '=ilike', v), ('name', '=ilike', v)], limit=1)

    @staticmethod
    def _bump(summary, created, key):
        bucket = 'created' if created else 'updated'
        summary[bucket][key] = summary[bucket].get(key, 0) + 1

    @api.model
    def _import_user(self, service, urow):
        Link = self.env['fusion.billing.account.link']
        ext = str(urow['id'])
        email = (urow.get('billing_email') or urow.get('email') or '').strip().lower() or None
        name = urow.get('full_name') or urow.get('company') or email or ext
        existed = bool(Link.search(
            [('service_id', '=', service.id), ('external_id', '=', ext)], limit=1))
        link = Link._resolve_or_create_partner(service, ext, name=name, email=email)
        vals = {}
        if urow.get('billing_address'):
            vals['street'] = urow['billing_address']
        if urow.get('billing_city'):
            vals['city'] = urow['billing_city']
        if urow.get('billing_postal_code'):
            vals['zip'] = urow['billing_postal_code']
        if urow.get('tax_id'):
            vals['vat'] = urow['tax_id']
        if urow.get('stripe_customer_id'):
            vals['x_fc_stripe_customer_id'] = urow['stripe_customer_id']
        country = self._fc_resolve_country(urow.get('billing_country'))
        if country:
            vals['country_id'] = country.id
        if vals:
            link.partner_id.write(vals)
        return link, not existed

    @api.model
    def _do_import(self, data):
        service = self._fc_service()
        summary = {'created': {}, 'updated': {}, 'skipped': [], 'failed': []}
        partner_by_user = {}
        for u in data.get('users', []):
            try:
                with self.env.cr.savepoint():
                    link, created = self._import_user(service, u)
                partner_by_user[str(u['id'])] = link.partner_id
                self._bump(summary, created, 'partners')
            except Exception as e:  # noqa: BLE001 - per-row isolation
                summary['failed'].append(
                    {'kind': 'user', 'id': str(u.get('id')), 'error': str(e)})
        return summary

Note: partner_by_user and (Task 3) plan_ctx_by_id are method-local dicts — never set them as attributes on self (Odoo recordsets reject arbitrary attribute assignment). Tasks 3 and 4 add their loops to this same _do_import method, so the locals stay in scope.

• Step 4: Run it, expect pass

Run: bash scripts/fcb_test_on_trial.sh Expected: FCB_EXIT=0; TestImporterIdentity passes. If country_id.code assertion fails, fix _fc_resolve_country (don't weaken the assertion).

• Step 5: Commit

git add fusion_centralize_billing/wizards/import_wizard.py fusion_centralize_billing/tests/test_importer.py fusion_centralize_billing/tests/__init__.py
git commit -m "feat(billing): importer identity (NexaCloud users -> partners + links)"

---

Task 3: Catalog import (plans → metric + products + charge, plan_id NULL)

Files:

• Modify: fusion_centralize_billing/wizards/import_wizard.py

• Modify: fusion_centralize_billing/tests/test_importer.py

• Step 1: Write the failing test (append to test_importer.py)

@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")
        self.assertTrue(charge.product_id.recurring_invoice is False
                        or charge.product_id.recurring_invoice in (False, None))

    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)

• Step 2: Run it, expect failure

Run: bash scripts/fcb_test_on_trial.sh Expected: FAIL — no charge created (catalog import not implemented).

• Step 3: Implement catalog import

Add to wizards/import_wizard.py:

    @api.model
    def _import_plan(self, metric, prow):
        Product = self.env['product.product']
        Charge = self.env['fusion.billing.charge']
        plan_code = str(prow['id'])
        name = prow.get('name') or plan_code
        price_monthly = float(prow.get('price_monthly') or 0.0)
        price_yearly = float(prow.get('price_yearly') or 0.0)

        sub_code = 'NC-PLAN-%s' % plan_code
        sub_product = Product.search([('default_code', '=', sub_code)], limit=1)
        created = False
        if not sub_product:
            sub_product = Product.create({
                'name': 'NexaCloud %s' % name, 'default_code': sub_code,
                'type': 'service', 'recurring_invoice': True,
                'list_price': price_monthly})
            created = True

        ov_code = 'NC-CPU-OVG-%s' % plan_code
        ov_product = Product.search([('default_code', '=', ov_code)], limit=1)
        if not ov_product:
            ov_product = Product.create({
                'name': 'NexaCloud CPU overage (%s)' % name, 'default_code': ov_code,
                'type': 'service', 'list_price': 0.0})

        charge_vals = {
            'name': 'NexaCloud CPU overage — %s' % name,
            'plan_code': plan_code, 'metric_id': metric.id, 'product_id': ov_product.id,
            'included_quota': float(prow.get('cpu_seconds_quota') or 0.0),
            'price_per_unit': CPU_RATE_PER_CORE_HOUR, 'unit_batch': CPU_SECONDS_PER_CORE_HOUR,
            'charge_model': 'standard',
            # plan_id intentionally omitted (NULL) — shadow safety guarantee #3
        }
        charge = Charge.search(
            [('plan_code', '=', plan_code), ('metric_id', '=', metric.id)], limit=1)
        if charge:
            charge.write(charge_vals)
        else:
            charge = Charge.create(charge_vals)
            created = True
        return {'sub_product': sub_product, 'overage_product': ov_product,
                'charge': charge, 'price_monthly': price_monthly,
                'price_yearly': price_yearly}, created

In _do_import, after the users loop, add the plans loop:

        metric = self._fc_cpu_metric()
        plan_ctx_by_id = {}
        for p in data.get('plans', []):
            try:
                with self.env.cr.savepoint():
                    ctx, created = self._import_plan(metric, p)
                plan_ctx_by_id[str(p['id'])] = ctx
                self._bump(summary, created, 'plans')
            except Exception as e:  # noqa: BLE001
                summary['failed'].append(
                    {'kind': 'plan', 'id': str(p.get('id')), 'error': str(e)})

• Step 4: Run it, expect pass

Run: bash scripts/fcb_test_on_trial.sh Expected: FCB_EXIT=0; both catalog tests pass. If product.product rejects recurring_invoice or type='service', read the field on odoo-trial and fix the source.

• Step 5: Commit

git add fusion_centralize_billing/wizards/import_wizard.py fusion_centralize_billing/tests/test_importer.py
git commit -m "feat(billing): importer catalog (plans -> products + CPU charge, plan_id NULL)"

---

Task 4: Subscription import (deployments → draft shadow sale.order)

Files:

• Modify: fusion_centralize_billing/wizards/import_wizard.py

• Modify: fusion_centralize_billing/tests/test_importer.py

• Step 1: Write the failing test (append to test_importer.py)

@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
        # monthly flat price set explicitly on the plan product line
        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
        # the yearly subscription gets the yearly price + yearly recurrence
        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']))

• Step 2: Run it, expect failure

Run: bash scripts/fcb_test_on_trial.sh Expected: FAIL — no subscriptions created (subscription import not implemented).

• Step 3: Implement subscription import

Add to wizards/import_wizard.py:

    @api.model
    def _import_subscription(self, service, partner, plan_ctx, recurrence_plans, srow):
        SaleOrder = self.env['sale.order']
        SaleOrderLine = self.env['sale.order.line']
        sub_ext = str(srow['id'])
        cycle = (srow.get('billing_cycle') or 'monthly').lower()
        rec_plan = recurrence_plans['yearly'] if cycle == 'yearly' else recurrence_plans['monthly']
        price = plan_ctx['price_yearly'] if cycle == 'yearly' else plan_ctx['price_monthly']
        product = plan_ctx['sub_product']
        order_vals = {
            'partner_id': partner.id, 'plan_id': rec_plan.id,
            'x_fc_nexacloud_subscription_id': sub_ext,
            'x_fc_nexacloud_deployment_id': str(srow.get('deployment_id') or ''),
            'x_fc_billing_service_id': service.id, 'x_fc_shadow': True,
        }
        existing = SaleOrder.search(
            [('x_fc_nexacloud_subscription_id', '=', sub_ext)], limit=1)
        if existing:
            existing.write(order_vals)
            line = existing.order_line.filtered(lambda l: l.product_id == product)
            line_vals = {'product_uom_qty': 1, 'price_unit': price}
            if line:
                line.write(line_vals)
            else:
                SaleOrderLine.create(dict(order_id=existing.id, product_id=product.id, **line_vals))
            order = existing
            created = False
        else:
            order_vals['order_line'] = [(0, 0, {
                'product_id': product.id, 'product_uom_qty': 1, 'price_unit': price})]
            order = SaleOrder.create(order_vals)
            created = True
        # guarantee the explicit price stuck (a pricelist compute may have overwritten it)
        line = order.order_line.filtered(lambda l: l.product_id == product)
        if line and line.price_unit != price:
            line.price_unit = price
        return order, created

In _do_import, before return summary, add the recurrences + subscriptions loop:

        recurrence_plans = {'monthly': self._fc_recurrence_plan('month'),
                            'yearly': self._fc_recurrence_plan('year')}
        for s in data.get('subscriptions', []):
            partner = partner_by_user.get(str(s.get('user_id') or ''))
            ctx = plan_ctx_by_id.get(str(s.get('plan_id') or ''))
            if not partner or not ctx:
                summary['skipped'].append({
                    'kind': 'subscription', 'id': str(s.get('id')),
                    'reason': 'unresolved %s' % ('user' if not partner else 'plan')})
                continue
            try:
                with self.env.cr.savepoint():
                    _order, created = self._import_subscription(
                        service, partner, ctx, recurrence_plans, s)
                self._bump(summary, created, 'subscriptions')
            except Exception as e:  # noqa: BLE001
                summary['failed'].append(
                    {'kind': 'subscription', 'id': str(s.get('id')), 'error': str(e)})

• Step 4: Run it, expect pass

Run: bash scripts/fcb_test_on_trial.sh Expected: FCB_EXIT=0. If is_subscription is False on the draft order, that disproves the design assumption — read sale_order.py in sale_subscription on odoo-trial and adjust how the subscription is created (e.g. set the field driving is_subscription), never weaken the assertion. If billing_period_unit rejects 'year', read the selection values and fix _fc_recurrence_plan.

• Step 5: Commit

git add fusion_centralize_billing/wizards/import_wizard.py fusion_centralize_billing/tests/test_importer.py
git commit -m "feat(billing): importer subscriptions (one draft shadow sale.order per deployment)"

---

Task 5: Idempotency + dry-run

Files:

• Modify: fusion_centralize_billing/tests/test_importer.py

• Step 1: Write the failing test (append to test_importer.py)

@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()
        # change a value and re-run; counts stay the same, value updates
        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")
        # the nexacloud service is created inside the rolled-back savepoint too
        self.assertFalse(self.env['fusion.billing.service'].search([('code', '=', 'nexacloud')]))

• Step 2: Run it, expect pass

Run: bash scripts/fcb_test_on_trial.sh Expected: FCB_EXIT=0 — idempotency and dry-run already hold from Tasks 2–4 + the savepoint in _import_rows. If the dry-run leaves a nexacloud service behind, the savepoint isn't wrapping _fc_service — confirm _do_import (which creates the service) runs entirely inside the with self.env.cr.savepoint() block.

• Step 3: Commit

git add fusion_centralize_billing/tests/test_importer.py
git commit -m "test(billing): importer idempotency + dry-run"

---

Task 6: Shadow-mode safety assertions

Files:

• Modify: fusion_centralize_billing/tests/test_importer.py

• Step 1: Write the failing test (append to test_importer.py)

@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')
        # no posted/draft customer invoice for any imported partner
        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")
        # no Stripe payment token -> charging is physically impossible
        tokens = self.env['payment.token'].search([('partner_id', 'in', partners.ids)])
        self.assertFalse(tokens, "shadow import must not attach a payment token")
        # every imported charge has a NULL plan_id so the rating cron skips it
        charges = self.env['fusion.billing.charge'].search([('plan_code', 'like', 'p-%')])
        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")

• Step 2: Run it, expect pass

Run: bash scripts/fcb_test_on_trial.sh Expected: FCB_EXIT=0 — the safety properties hold by construction (draft, no token, NULL plan_id). If payment.token is not a valid model name in this build, read the payment model names on odoo-trial and use the correct one (don't drop the assertion). If an invoice is found, the draft-import guarantee is broken — investigate whether sale.order.create auto-invoices, and stop confirming/posting.

• Step 3: Commit

git add fusion_centralize_billing/tests/test_importer.py
git commit -m "test(billing): importer shadow-mode safety (no invoice/token, cron no-op)"

---

Task 7: Error handling — malformed rows isolated

Files:

• Modify: fusion_centralize_billing/tests/test_importer.py

• Step 1: Write the failing test (append to test_importer.py)

@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)
        # the two good users still import
        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']))

• Step 2: Run it, expect pass

Run: bash scripts/fcb_test_on_trial.sh Expected: FCB_EXIT=0 — the per-row try/except + savepoint already isolates failures. If the whole batch aborts, the savepoint is missing around _import_user or the broad except is too narrow — fix so one bad row never poisons the cursor.

• Step 3: Commit

git add fusion_centralize_billing/tests/test_importer.py
git commit -m "test(billing): importer per-row error isolation"

---

Task 8: Read path — DSN guard

Files:

• Modify: fusion_centralize_billing/tests/test_importer.py

• Step 1: Write the failing test (append to test_importer.py)

from odoo.exceptions import UserError


@tagged('post_install', '-at_install')
class TestImporterReadGuard(TransactionCase):

    def test_missing_dsn_raises_usererror(self):
        # ensure no DSN is configured in the test DB
        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()

• Step 2: Run it, expect pass

Run: bash scripts/fcb_test_on_trial.sh Expected: FCB_EXIT=0 — _read_nexacloud_rows raises UserError when the DSN param is empty (implemented in Task 1). If psycopg2 import fails on odoo-trial, confirm it ships with the image (it does — Odoo depends on it).

• Step 3: Commit

git add fusion_centralize_billing/tests/test_importer.py
git commit -m "test(billing): importer read-path DSN guard"

---

Task 9: Full suite + static checks

Files: none (verification task)

• Step 1: Full test run

Run: bash scripts/fcb_test_on_trial.sh Expected: FCB_EXIT=0, no FAIL/ERROR lines for fusion_centralize_billing.

• Step 2: No _sql_constraints regressions

Run: grep -rn "_sql_constraints" fusion_centralize_billing/ || echo "clean" Expected: clean.

• Step 3: No bare sale.subscription model references

Run: grep -rnE "sale\.subscription[^.]" fusion_centralize_billing/ || echo "clean" Expected: clean (only sale.subscription.plan is valid).

• Step 4: Pyflakes the new Python

Run: docker exec odoo-modsdev-app python3 -m pyflakes fusion_centralize_billing/wizards/import_wizard.py fusion_centralize_billing/models/res_partner.py 2>&1 | tail -20 || true Expected: no undefined names (catches the kind of _norm_email NameError the helpdesk smoke test missed).

• Step 5: Commit (if any fixes)

git add -A fusion_centralize_billing/
git commit -m "test(billing): 2a importer full suite green + static checks"

---

Done = 2a importer complete

A NexaCloud backfill produces, idempotently: unified partners + links, a cpu_seconds charge catalog (plan_id NULL), and one draft shadow sale.order per deployment carrying the exact NexaCloud flat price — with zero customer-visible billing in Odoo (no invoice, no token, rating cron a no-op). The psycopg2 read path is ready; the live run is gated only on the read-only DSN grant.

Next (not this plan)

• 2b: NexaCloud usage_metering.py pushes cpu-seconds (= core-hours × 3600) to POST /usage.
• 2c: NexaCloud consumes invoice.payment_failed / subscription.terminated webhooks → throttle/deprovision.
• 2d: fusion.billing.reconciliation diffs Odoo-computed (flat + charge._compute_billable) vs NexaCloud actuals per period; flip when within tolerance (set charge.plan_id, attach tokens, confirm subs).