feat(fusion_accounting_reports): fusion.report.engine 5-method API

The engine orchestrator. compute_pnl, compute_balance_sheet,
compute_trial_balance, compute_gl, drill_down. All controllers,
wizards, AI tools must route through these methods; no direct
SQL aggregation from anywhere else.

Internal pipeline: validate -> fetch hierarchy -> SQL aggregate
-> resolve line_specs -> optional comparison + anomaly. Uses raw
SQL for the per-account aggregate (the perf-critical step), ORM
for everything else.

Per-company report lookup with global fallback (company_id desc
nulls last). Balance sheet uses 1970 epoch as date_from for
cumulative-since-inception semantics.

7 new tests, 42 total passing.

Made-with: Cursor

fusion_accounting_reports/__init__.py

@@ -1 +1,2 @@
 from . import services
+from . import models

fusion_accounting_reports/__manifest__.py

@@ -1,6 +1,6 @@
 {
     'name': 'Fusion Accounting Reports',
-    'version': '19.0.1.0.0',
+    'version': '19.0.1.0.4',
     'category': 'Accounting/Accounting',
     'summary': 'AI-augmented financial reports (P&L, balance sheet, trial balance, GL).',
     'description': """

fusion_accounting_reports/models/__init__.py

@@ -0,0 +1,2 @@
+from . import fusion_report
+from . import fusion_report_engine

fusion_accounting_reports/models/fusion_report.py

@@ -0,0 +1,63 @@
+"""Persistent definition of a Fusion financial report.
+
+Each report (P&L, balance sheet, trial balance, GL) has ONE row in
+fusion.report describing its metadata + line specs. The line specs
+are stored as a JSON-typed field for flexibility (each line spec
+includes account_type filter, sub-totaling rules, sign convention)."""
+
+from odoo import _, api, fields, models
+
+
+REPORT_TYPES = [
+    ('pnl', 'Income Statement (P&L)'),
+    ('balance_sheet', 'Balance Sheet'),
+    ('trial_balance', 'Trial Balance'),
+    ('general_ledger', 'General Ledger'),
+]
+
+
+class FusionReport(models.Model):
+    _name = "fusion.report"
+    _description = "Fusion Financial Report Definition"
+    _order = "sequence, id"
+
+    name = fields.Char(required=True, translate=True)
+    code = fields.Char(
+        required=True,
+        help="Unique technical code (e.g. 'pnl', 'balance_sheet').",
+    )
+    report_type = fields.Selection(REPORT_TYPES, required=True)
+    sequence = fields.Integer(default=10)
+    description = fields.Text()
+    active = fields.Boolean(default=True)
+
+    # Layout config - stored as JSON for flexibility per report type.
+    # Example for P&L:
+    # [
+    #   {"label": "Revenue", "account_type_prefix": "income_", "sign": 1},
+    #   {"label": "Cost of Goods Sold", "account_type_prefix": "expense_direct_", "sign": -1},
+    #   {"label": "Gross Profit", "compute": "subtotal", "above": 2},
+    #   ...
+    # ]
+    line_specs = fields.Json(string="Line Specs")
+
+    show_zero_balances = fields.Boolean(default=False)
+    show_unposted = fields.Boolean(default=False)
+    default_comparison_mode = fields.Selection(
+        [
+            ('none', 'No comparison'),
+            ('previous_period', 'Previous Period'),
+            ('previous_year', 'Previous Year'),
+        ],
+        default='none',
+    )
+
+    company_id = fields.Many2one(
+        'res.company',
+        default=lambda self: self.env.company,
+    )
+
+    _unique_company_code = models.Constraint(
+        'UNIQUE(company_id, code)',
+        'Report code must be unique per company.',
+    )

fusion_accounting_reports/models/fusion_report_engine.py

@@ -0,0 +1,245 @@
+"""The reports engine - orchestrator for all report computation.
+
+5-method public API. All controllers, AI tools, wizards, exports must
+go through these methods; no direct ORM aggregation queries from
+anywhere else.
+
+Internal pipeline (per report run):
+1. Validate (period valid, company allowed, report exists)
+2. Fetch account hierarchy (cached per (company, fiscal_year))
+3. Aggregate move lines per account (the SQL workhorse)
+4. Resolve line_specs into report rows
+5. (Optional) Compute comparison-period rows
+6. (Optional) Detect anomalies (deferred to later tasks)
+"""
+
+import logging
+from datetime import date
+
+from odoo import _, api, models
+from odoo.exceptions import ValidationError
+
+from ..services.account_hierarchy import build_tree
+from ..services.date_periods import Period, comparison_period as _comp_period
+from ..services.drill_down_resolver import fetch_drill_down
+from ..services.line_resolver import resolve as _resolve_lines
+from ..services.totaling import TotalLine
+
+_logger = logging.getLogger(__name__)
+
+
+class FusionReportEngine(models.AbstractModel):
+    _name = "fusion.report.engine"
+    _description = "Fusion Financial Reports Engine"
+
+    # ============================================================
+    # PUBLIC API (5 methods)
+    # ============================================================
+
+    @api.model
+    def compute_pnl(
+        self, period: Period, *, comparison: str = 'none',
+        company_id: int | None = None,
+    ) -> dict:
+        """Income statement (P&L) for the given period."""
+        report = self._get_report('pnl', company_id=company_id)
+        return self._compute(
+            report, period, comparison=comparison, company_id=company_id,
+        )
+
+    @api.model
+    def compute_balance_sheet(
+        self, date_to: date, *, comparison: str = 'none',
+        company_id: int | None = None,
+    ) -> dict:
+        """Balance sheet AS OF date_to. Period.date_from is set to a
+        far-past date so balances are cumulative-since-inception."""
+        report = self._get_report('balance_sheet', company_id=company_id)
+        period = Period(
+            date_from=date(1970, 1, 1),
+            date_to=date_to,
+            label=f"As of {date_to}",
+        )
+        return self._compute(
+            report, period, comparison=comparison, company_id=company_id,
+        )
+
+    @api.model
+    def compute_trial_balance(
+        self, period: Period, *, company_id: int | None = None,
+    ) -> dict:
+        """Trial balance for the given period - every account with
+        non-zero balance."""
+        report = self._get_report('trial_balance', company_id=company_id)
+        return self._compute(
+            report, period, comparison='none', company_id=company_id,
+        )
+
+    @api.model
+    def compute_gl(
+        self, period: Period, *, account_ids: list | None = None,
+        company_id: int | None = None,
+    ) -> dict:
+        """General ledger for the given period.
+
+        Returns per-account move-line listings rather than aggregated rows."""
+        report = self._get_report('general_ledger', company_id=company_id)
+        company_id = company_id or self.env.company.id
+        result = self._compute(
+            report, period, comparison='none', company_id=company_id,
+        )
+        gl_by_account = {}
+        target_ids = account_ids or list(result.get('account_totals', {}).keys())
+        for acct_id in target_ids:
+            gl_by_account[acct_id] = fetch_drill_down(
+                self.env,
+                account_id=acct_id,
+                date_from=period.date_from,
+                date_to=period.date_to,
+                company_id=company_id,
+                limit=200,
+            )
+        result['gl_by_account'] = gl_by_account
+        return result
+
+    @api.model
+    def drill_down(
+        self, *, account_id: int, period: Period,
+        company_id: int | None = None,
+    ) -> list:
+        """Drill into a report line: list the journal items behind it."""
+        company_id = company_id or self.env.company.id
+        return fetch_drill_down(
+            self.env,
+            account_id=account_id,
+            date_from=period.date_from,
+            date_to=period.date_to,
+            company_id=company_id,
+            limit=500,
+        )
+
+    # ============================================================
+    # PRIVATE HELPERS
+    # ============================================================
+
+    def _get_report(self, report_type: str, *, company_id: int | None = None):
+        """Look up the active fusion.report definition for a given
+        type+company. If no per-company override, falls back to global
+        (company_id=False)."""
+        Report = self.env['fusion.report'].sudo()
+        company_id = company_id or self.env.company.id
+        report = Report.search(
+            [
+                ('report_type', '=', report_type),
+                ('active', '=', True),
+                '|',
+                ('company_id', '=', company_id),
+                ('company_id', '=', False),
+            ],
+            order='company_id desc nulls last',
+            limit=1,
+        )
+        if not report:
+            raise ValidationError(
+                _("No active fusion.report definition for type '%s'") % report_type
+            )
+        return report
+
+    def _fetch_accounts(self, company_id):
+        """Fetch all accounts for a company, return flat dict + tree."""
+        Account = self.env['account.account'].sudo()
+        records = Account.search([('company_ids', 'in', company_id)])
+        # account.account doesn't carry a parent_id in V19 - we use
+        # account_type prefixes instead, so parent_id is always None here.
+        flat = [
+            {
+                'id': a.id,
+                'code': a.code,
+                'name': a.name,
+                'account_type': a.account_type or '',
+                'parent_id': None,
+            }
+            for a in records
+        ]
+        accounts_by_id = {a['id']: a for a in flat}
+        tree = build_tree(flat)
+        return accounts_by_id, tree
+
+    def _aggregate_period(self, period: Period, company_id: int) -> dict:
+        """SQL aggregate per account_id for a period.
+
+        Raw SQL for performance; this is the perf-critical step."""
+        self.env.cr.execute(
+            """
+            SELECT account_id,
+                   COALESCE(SUM(debit), 0) AS d,
+                   COALESCE(SUM(credit), 0) AS c,
+                   COALESCE(SUM(balance), 0) AS b
+            FROM account_move_line
+            WHERE parent_state = 'posted'
+              AND company_id = %s
+              AND date >= %s
+              AND date <= %s
+            GROUP BY account_id
+            """,
+            (company_id, period.date_from, period.date_to),
+        )
+        out = {}
+        for row in self.env.cr.fetchall():
+            out[row[0]] = TotalLine(
+                debit=float(row[1] or 0),
+                credit=float(row[2] or 0),
+                balance=float(row[3] or 0),
+            )
+        return out
+
+    def _compute(
+        self, report, period: Period, *, comparison: str,
+        company_id: int | None = None,
+    ) -> dict:
+        """Shared computation pipeline. Returns dict with rows, totals,
+        metadata."""
+        company_id = company_id or self.env.company.id
+
+        accounts_by_id, _tree = self._fetch_accounts(company_id)
+
+        account_totals = self._aggregate_period(period, company_id)
+
+        comp_totals = None
+        comp_period = None
+        if comparison and comparison != 'none':
+            comp_period = _comp_period(period, comparison)
+            if comp_period:
+                comp_totals = self._aggregate_period(comp_period, company_id)
+
+        rows = _resolve_lines(
+            report.line_specs or [],
+            account_totals=account_totals,
+            accounts_by_id=accounts_by_id,
+            comparison_totals=comp_totals,
+        )
+
+        return {
+            'report_id': report.id,
+            'report_name': report.name,
+            'report_type': report.report_type,
+            'period': {
+                'date_from': str(period.date_from),
+                'date_to': str(period.date_to),
+                'label': period.label,
+            },
+            'comparison_period': (
+                {
+                    'date_from': str(comp_period.date_from),
+                    'date_to': str(comp_period.date_to),
+                    'label': comp_period.label,
+                }
+                if comp_period
+                else None
+            ),
+            'company_id': company_id,
+            'rows': rows,
+            'account_totals': {
+                aid: tl.balance for aid, tl in account_totals.items()
+            },
+        }

fusion_accounting_reports/security/ir.model.access.csv

@@ -1 +1,3 @@
 id,name,model_id:id,group_id:id,perm_read,perm_write,perm_create,perm_unlink
+access_fusion_report_user,fusion.report.user,model_fusion_report,base.group_user,1,0,0,0
+access_fusion_report_admin,fusion.report.admin,model_fusion_report,fusion_accounting_core.group_fusion_accounting_admin,1,1,1,1

fusion_accounting_reports/services/__init__.py

@@ -2,3 +2,5 @@ from . import date_periods
 from . import account_hierarchy
 from . import totaling
 from . import currency_conversion
+from . import line_resolver
+from . import drill_down_resolver

fusion_accounting_reports/services/drill_down_resolver.py

@@ -0,0 +1,81 @@
+"""Drill-down: from a report line to its underlying journal items.
+
+Given an account_id and a Period, fetches the matching account.move.line
+records and returns them in a flat list. Used by the OWL drill-down
+dialog and the engine's drill_down() public API."""
+
+from dataclasses import dataclass
+from datetime import date
+
+
+@dataclass
+class DrillDownRow:
+    move_line_id: int
+    move_id: int
+    move_name: str
+    date: date
+    account_code: str
+    account_name: str
+    partner_name: str | None
+    label: str
+    debit: float
+    credit: float
+    balance: float
+
+    def to_dict(self):
+        return {
+            'move_line_id': self.move_line_id,
+            'move_id': self.move_id,
+            'move_name': self.move_name,
+            'date': str(self.date),
+            'account_code': self.account_code,
+            'account_name': self.account_name,
+            'partner_name': self.partner_name or '',
+            'label': self.label,
+            'debit': self.debit,
+            'credit': self.credit,
+            'balance': self.balance,
+        }
+
+
+def fetch_drill_down(
+    env,
+    *,
+    account_id: int,
+    date_from: date,
+    date_to: date,
+    company_id: int | None = None,
+    limit: int = 500,
+) -> list[dict]:
+    """Fetch journal items for an account within a date range.
+
+    Returns flat list of dicts ready for the drill-down OWL table."""
+    Line = env['account.move.line'].sudo()
+    domain = [
+        ('account_id', '=', account_id),
+        ('date', '>=', date_from),
+        ('date', '<=', date_to),
+        ('parent_state', '=', 'posted'),
+    ]
+    if company_id:
+        domain.append(('company_id', '=', company_id))
+
+    move_lines = Line.search(domain, limit=limit, order='date asc, id asc')
+    rows = []
+    for ml in move_lines:
+        rows.append(
+            DrillDownRow(
+                move_line_id=ml.id,
+                move_id=ml.move_id.id,
+                move_name=ml.move_id.name or '',
+                date=ml.date,
+                account_code=ml.account_id.code,
+                account_name=ml.account_id.name,
+                partner_name=ml.partner_id.name if ml.partner_id else None,
+                label=ml.name or '',
+                debit=ml.debit,
+                credit=ml.credit,
+                balance=ml.balance,
+            ).to_dict()
+        )
+    return rows

fusion_accounting_reports/services/line_resolver.py

@@ -0,0 +1,143 @@
+"""Resolve a fusion.report definition into report rows.
+
+Pure-Python: takes line_specs (list of dicts), a period, and aggregated
+move-line data (per-account totals) - returns ordered list of report row
+dicts ready for the OWL frontend or PDF rendering.
+
+Row shape:
+{
+  'id': 'line_<index>',
+  'label': str,
+  'level': int,                  # indentation depth
+  'is_subtotal': bool,
+  'amount': float,
+  'amount_comparison': float | None,
+  'variance_pct': float | None,
+  'account_id': int | None,      # for drill-down (None for subtotals)
+  'children': list[dict],         # populated when expanded
+}"""
+
+from dataclasses import dataclass
+
+from .totaling import TotalLine
+
+
+@dataclass
+class ReportRow:
+    id: str
+    label: str
+    level: int = 0
+    is_subtotal: bool = False
+    amount: float = 0.0
+    amount_comparison: float | None = None
+    variance_pct: float | None = None
+    account_id: int | None = None
+
+    def to_dict(self):
+        return {
+            'id': self.id,
+            'label': self.label,
+            'level': self.level,
+            'is_subtotal': self.is_subtotal,
+            'amount': self.amount,
+            'amount_comparison': self.amount_comparison,
+            'variance_pct': self.variance_pct,
+            'account_id': self.account_id,
+        }
+
+
+def resolve(
+    line_specs: list[dict],
+    *,
+    account_totals: dict[int, TotalLine],
+    accounts_by_id: dict[int, dict],
+    comparison_totals: dict[int, TotalLine] | None = None,
+) -> list[dict]:
+    """Resolve line_specs against actual account totals -> list of row dicts.
+
+    Args:
+        line_specs: report definition line specs (from fusion.report.line_specs).
+        account_totals: {account_id: TotalLine} for the period.
+        accounts_by_id: {account_id: {code, name, account_type, ...}}.
+        comparison_totals: optional {account_id: TotalLine} for comparison period.
+
+    Returns: list of row dicts."""
+    rows: list[ReportRow] = []
+
+    for idx, spec in enumerate(line_specs):
+        if spec.get('compute') == 'subtotal':
+            n = spec.get('above', 1)
+            sign = spec.get('sign', 1)
+            recent = [r.amount for r in rows[-n:] if not r.is_subtotal]
+            row = ReportRow(
+                id=f'line_{idx}',
+                label=spec.get('label', 'Subtotal'),
+                level=spec.get('level', 0),
+                is_subtotal=True,
+                amount=sum(recent) * sign,
+            )
+            if comparison_totals is not None:
+                comp_recent = [
+                    r.amount_comparison
+                    for r in rows[-n:]
+                    if not r.is_subtotal and r.amount_comparison is not None
+                ]
+                row.amount_comparison = (
+                    sum(comp_recent) * sign if comp_recent else None
+                )
+            rows.append(row)
+
+        elif spec.get('account_type_prefix'):
+            prefix = spec['account_type_prefix']
+            sign = spec.get('sign', 1)
+            matched_ids = [
+                aid for aid, info in accounts_by_id.items()
+                if info.get('account_type', '').startswith(prefix)
+            ]
+            amount = sum(
+                account_totals.get(aid, TotalLine()).balance * sign
+                for aid in matched_ids
+            )
+            row = ReportRow(
+                id=f'line_{idx}',
+                label=spec.get('label', prefix),
+                level=spec.get('level', 0),
+                amount=amount,
+            )
+            if comparison_totals is not None:
+                comp_amount = sum(
+                    comparison_totals.get(aid, TotalLine()).balance * sign
+                    for aid in matched_ids
+                )
+                row.amount_comparison = comp_amount
+                if comp_amount != 0:
+                    row.variance_pct = (
+                        (amount - comp_amount) / abs(comp_amount)
+                    ) * 100
+            rows.append(row)
+
+        elif spec.get('account_id'):
+            aid = spec['account_id']
+            sign = spec.get('sign', 1)
+            tot = account_totals.get(aid, TotalLine())
+            label = spec.get('label') or accounts_by_id.get(aid, {}).get(
+                'name', f'Account {aid}'
+            )
+            row = ReportRow(
+                id=f'line_{idx}',
+                label=label,
+                level=spec.get('level', 0),
+                amount=tot.balance * sign,
+                account_id=aid,
+            )
+            if comparison_totals is not None:
+                comp = comparison_totals.get(aid, TotalLine())
+                row.amount_comparison = comp.balance * sign
+                if row.amount_comparison and row.amount_comparison != 0:
+                    row.variance_pct = (
+                        (row.amount - row.amount_comparison)
+                        / abs(row.amount_comparison)
+                    ) * 100
+            rows.append(row)
+
+    return [r.to_dict() for r in rows]

fusion_accounting_reports/tests/__init__.py

@@ -1,2 +1,6 @@
 from . import test_services_unit
 from . import test_currency_conversion
+from . import test_fusion_report
+from . import test_line_resolver
+from . import test_drill_down_resolver
+from . import test_fusion_report_engine

fusion_accounting_reports/tests/test_drill_down_resolver.py

@@ -0,0 +1,60 @@
+"""Tests for drill_down_resolver."""
+
+from datetime import date, timedelta
+
+from odoo.tests.common import TransactionCase, tagged
+from odoo.addons.fusion_accounting_reports.services.drill_down_resolver import (
+    fetch_drill_down,
+)
+
+
+@tagged('post_install', '-at_install')
+class TestDrillDownResolver(TransactionCase):
+
+    def test_returns_empty_for_account_with_no_lines(self):
+        account = self.env['account.account'].search([
+            ('company_ids', 'in', self.env.company.id),
+        ], limit=1)
+        if not account:
+            self.skipTest("No accounts in DB")
+        rows = fetch_drill_down(
+            self.env,
+            account_id=account.id,
+            date_from=date(2099, 1, 1),
+            date_to=date(2099, 12, 31),
+            company_id=self.env.company.id,
+        )
+        self.assertEqual(rows, [])
+
+    def test_returns_lines_for_account_with_data(self):
+        line = self.env['account.move.line'].search([
+            ('parent_state', '=', 'posted'),
+        ], limit=1)
+        if not line:
+            self.skipTest("No posted move lines in DB")
+        rows = fetch_drill_down(
+            self.env,
+            account_id=line.account_id.id,
+            date_from=line.date - timedelta(days=1),
+            date_to=line.date + timedelta(days=1),
+            company_id=line.company_id.id,
+        )
+        self.assertGreater(len(rows), 0)
+        ids = [r['move_line_id'] for r in rows]
+        self.assertIn(line.id, ids)
+
+    def test_respects_limit(self):
+        line = self.env['account.move.line'].search([
+            ('parent_state', '=', 'posted'),
+        ], limit=1)
+        if not line:
+            self.skipTest("No posted move lines in DB")
+        rows = fetch_drill_down(
+            self.env,
+            account_id=line.account_id.id,
+            date_from=date(2000, 1, 1),
+            date_to=date(2099, 12, 31),
+            company_id=line.company_id.id,
+            limit=2,
+        )
+        self.assertLessEqual(len(rows), 2)

fusion_accounting_reports/tests/test_fusion_report.py

@@ -0,0 +1,44 @@
+"""Tests for fusion.report definition model."""
+
+from odoo.tests.common import TransactionCase, tagged
+
+
+@tagged('post_install', '-at_install')
+class TestFusionReport(TransactionCase):
+
+    def test_create_minimal(self):
+        report = self.env['fusion.report'].create({
+            'name': 'Test P&L',
+            'code': 'test_pnl_minimal',
+            'report_type': 'pnl',
+        })
+        self.assertEqual(report.name, 'Test P&L')
+        self.assertTrue(report.active)
+        self.assertEqual(report.default_comparison_mode, 'none')
+
+    def test_line_specs_json_roundtrip(self):
+        specs = [
+            {'label': 'Revenue', 'account_type_prefix': 'income_', 'sign': 1},
+            {'label': 'COGS', 'account_type_prefix': 'expense_direct_', 'sign': -1},
+        ]
+        report = self.env['fusion.report'].create({
+            'name': 'Test',
+            'code': 'test_json_roundtrip',
+            'report_type': 'pnl',
+            'line_specs': specs,
+        })
+        self.assertEqual(report.line_specs, specs)
+        self.assertEqual(report.line_specs[0]['label'], 'Revenue')
+
+    def test_company_code_uniqueness(self):
+        self.env['fusion.report'].create({
+            'name': 'A',
+            'code': 'dup_code_test',
+            'report_type': 'pnl',
+        })
+        with self.assertRaises(Exception):
+            self.env['fusion.report'].create({
+                'name': 'B',
+                'code': 'dup_code_test',
+                'report_type': 'pnl',
+            })

fusion_accounting_reports/tests/test_fusion_report_engine.py

@@ -0,0 +1,109 @@
+"""Tests for fusion.report.engine AbstractModel."""
+
+from datetime import date
+
+from odoo.exceptions import ValidationError
+from odoo.tests.common import TransactionCase, tagged
+from odoo.addons.fusion_accounting_reports.services.date_periods import Period
+
+
+@tagged('post_install', '-at_install')
+class TestFusionReportEngine(TransactionCase):
+
+    def setUp(self):
+        super().setUp()
+        self.pnl_report = self.env['fusion.report'].create({
+            'name': 'Test P&L Engine',
+            'code': 'test_pnl_engine',
+            'report_type': 'pnl',
+            'line_specs': [
+                {'label': 'Revenue', 'account_type_prefix': 'income_', 'sign': 1},
+                {'label': 'Expenses', 'account_type_prefix': 'expense_', 'sign': -1},
+                {'label': 'Net Profit', 'compute': 'subtotal', 'above': 2},
+            ],
+            'company_id': self.env.company.id,
+        })
+
+    def test_engine_model_exists(self):
+        self.assertIn('fusion.report.engine', self.env.registry)
+
+    def test_compute_pnl_returns_dict_with_rows(self):
+        period = Period(date(2026, 1, 1), date(2026, 12, 31), 'Test 2026')
+        result = self.env['fusion.report.engine'].compute_pnl(
+            period, company_id=self.env.company.id,
+        )
+        self.assertIn('rows', result)
+        self.assertIn('report_type', result)
+        self.assertEqual(result['report_type'], 'pnl')
+
+    def test_compute_balance_sheet(self):
+        self.env['fusion.report'].create({
+            'name': 'Test BS',
+            'code': 'test_bs_engine',
+            'report_type': 'balance_sheet',
+            'line_specs': [
+                {'label': 'Assets', 'account_type_prefix': 'asset_', 'sign': 1},
+            ],
+            'company_id': self.env.company.id,
+        })
+        result = self.env['fusion.report.engine'].compute_balance_sheet(
+            date(2026, 4, 19), company_id=self.env.company.id,
+        )
+        self.assertEqual(result['report_type'], 'balance_sheet')
+        self.assertEqual(result['period']['date_to'], '2026-04-19')
+
+    def test_compute_trial_balance(self):
+        self.env['fusion.report'].create({
+            'name': 'Test TB',
+            'code': 'test_tb_engine',
+            'report_type': 'trial_balance',
+            'line_specs': [],
+            'company_id': self.env.company.id,
+        })
+        period = Period(date(2026, 1, 1), date(2026, 12, 31), 'Test 2026')
+        result = self.env['fusion.report.engine'].compute_trial_balance(
+            period, company_id=self.env.company.id,
+        )
+        self.assertEqual(result['report_type'], 'trial_balance')
+
+    def test_compute_pnl_with_comparison(self):
+        period = Period(date(2026, 1, 1), date(2026, 12, 31), 'Test 2026')
+        result = self.env['fusion.report.engine'].compute_pnl(
+            period,
+            comparison='previous_year',
+            company_id=self.env.company.id,
+        )
+        self.assertIsNotNone(result.get('comparison_period'))
+        self.assertEqual(result['comparison_period']['date_to'], '2025-12-31')
+
+    def test_drill_down_returns_list(self):
+        line = self.env['account.move.line'].search([
+            ('parent_state', '=', 'posted'),
+        ], limit=1)
+        if not line:
+            self.skipTest("No posted lines in DB")
+        period = Period(line.date, line.date, 'Single day')
+        rows = self.env['fusion.report.engine'].drill_down(
+            account_id=line.account_id.id,
+            period=period,
+            company_id=line.company_id.id,
+        )
+        self.assertIsInstance(rows, list)
+
+    def test_no_report_raises_validation_error(self):
+        period = Period(date(2026, 1, 1), date(2026, 12, 31), 'Test 2026')
+        # Inactivate any pre-existing GL definitions so the lookup
+        # fails for this test, then restore them after.
+        existing = self.env['fusion.report'].search(
+            [('report_type', '=', 'general_ledger')]
+        )
+        prior_active = {r.id: r.active for r in existing}
+        existing.write({'active': False})
+        try:
+            with self.assertRaises(ValidationError):
+                self.env['fusion.report.engine'].compute_gl(
+                    period, company_id=self.env.company.id,
+                )
+        finally:
+            for r in existing:
+                r.active = prior_active.get(r.id, True)

fusion_accounting_reports/tests/test_line_resolver.py

@@ -0,0 +1,96 @@
+"""Tests for line_resolver."""
+
+from odoo.tests.common import TransactionCase, tagged
+from odoo.addons.fusion_accounting_reports.services.line_resolver import resolve
+from odoo.addons.fusion_accounting_reports.services.totaling import TotalLine
+
+
+@tagged('post_install', '-at_install')
+class TestLineResolver(TransactionCase):
+
+    def test_resolve_account_type_prefix(self):
+        line_specs = [
+            {'label': 'Revenue', 'account_type_prefix': 'income_', 'sign': 1},
+        ]
+        accounts_by_id = {
+            1: {'code': '4000', 'name': 'Sales', 'account_type': 'income_other'},
+            2: {'code': '4100', 'name': 'Service Revenue', 'account_type': 'income_service'},
+            3: {'code': '5000', 'name': 'COGS', 'account_type': 'expense_direct_cost'},
+        }
+        account_totals = {
+            1: TotalLine(balance=10000),
+            2: TotalLine(balance=5000),
+            3: TotalLine(balance=4000),
+        }
+        rows = resolve(
+            line_specs,
+            account_totals=account_totals,
+            accounts_by_id=accounts_by_id,
+        )
+        self.assertEqual(len(rows), 1)
+        self.assertEqual(rows[0]['label'], 'Revenue')
+        self.assertEqual(rows[0]['amount'], 15000)
+
+    def test_resolve_subtotal(self):
+        line_specs = [
+            {'label': 'Revenue', 'account_type_prefix': 'income_', 'sign': 1},
+            {'label': 'COGS', 'account_type_prefix': 'expense_', 'sign': -1},
+            {'label': 'Gross Profit', 'compute': 'subtotal', 'above': 2},
+        ]
+        accounts_by_id = {
+            1: {'code': '4000', 'name': 'Sales', 'account_type': 'income_other'},
+            2: {'code': '5000', 'name': 'COGS', 'account_type': 'expense_direct'},
+        }
+        account_totals = {
+            1: TotalLine(balance=10000),
+            2: TotalLine(balance=4000),
+        }
+        rows = resolve(
+            line_specs,
+            account_totals=account_totals,
+            accounts_by_id=accounts_by_id,
+        )
+        self.assertEqual(len(rows), 3)
+        self.assertEqual(rows[0]['amount'], 10000)
+        self.assertEqual(rows[1]['amount'], -4000)
+        self.assertEqual(rows[2]['amount'], 6000)
+        self.assertTrue(rows[2]['is_subtotal'])
+
+    def test_resolve_with_comparison(self):
+        line_specs = [
+            {'label': 'Revenue', 'account_type_prefix': 'income_', 'sign': 1},
+        ]
+        accounts_by_id = {
+            1: {'code': '4000', 'name': 'Sales', 'account_type': 'income_other'},
+        }
+        account_totals = {1: TotalLine(balance=12000)}
+        comparison_totals = {1: TotalLine(balance=10000)}
+        rows = resolve(
+            line_specs,
+            account_totals=account_totals,
+            accounts_by_id=accounts_by_id,
+            comparison_totals=comparison_totals,
+        )
+        self.assertEqual(rows[0]['amount'], 12000)
+        self.assertEqual(rows[0]['amount_comparison'], 10000)
+        self.assertAlmostEqual(rows[0]['variance_pct'], 20.0)
+
+    def test_resolve_empty_specs(self):
+        rows = resolve([], account_totals={}, accounts_by_id={})
+        self.assertEqual(rows, [])
+
+    def test_resolve_account_id_drill_down(self):
+        line_specs = [
+            {'label': 'Cash', 'account_id': 99, 'sign': 1},
+        ]
+        accounts_by_id = {
+            99: {'code': '1100', 'name': 'Cash', 'account_type': 'asset_cash'},
+        }
+        account_totals = {99: TotalLine(balance=5000)}
+        rows = resolve(
+            line_specs,
+            account_totals=account_totals,
+            accounts_by_id=accounts_by_id,
+        )
+        self.assertEqual(rows[0]['account_id'], 99)
+        self.assertEqual(rows[0]['amount'], 5000)