@@ -0,0 +1,956 @@
# 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
```
- 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–
- **Never** run `--test-enable` against production `nexamain` .
## File structure (this plan)
```
__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` ):
``` python
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` :
``` python
# -*- 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` :
``` python
from . import import_wizard
```
`fusion_centralize_billing/wizards/import_wizard.py` :
``` python
# -*- 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
<?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` :
```
```
In `__manifest__.py` , add the view to `data` (after the cron):
``` python
" 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 **
``` bash
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` :
``` python
# -*- 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` :
``` python
@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 **
``` bash
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` )
``` python
@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` :
``` python
@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:
``` python
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 **
``` bash
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` )
``` python
@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` :
``` python
@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:
``` python
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 **
``` bash
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` )
``` python
@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– `_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 **
``` bash
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` )
``` python
@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 **
``` bash
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` )
``` python
@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 **
``` bash
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` )
``` python
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 **
``` bash
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) **
``` bash
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 × `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).
