feat(fusion_accounting_reports): 2 cron jobs (anomaly scan + MV refresh)

Made-with: Cursor

fusion_accounting_reports/__manifest__.py

@@ -1,6 +1,6 @@
 {
     'name': 'Fusion Accounting Reports',
-    'version': '19.0.1.0.16',
+    'version': '19.0.1.0.21',
     'category': 'Accounting/Accounting',
     'summary': 'AI-augmented financial reports (P&L, balance sheet, trial balance, GL).',
     'description': """
@@ -35,6 +35,7 @@ menu hides; the engine and AI tools remain available for the chat.
         'data/report_balance_sheet.xml',
         'data/report_trial_balance.xml',
         'data/report_general_ledger.xml',
+        'data/cron.xml',
     ],
     'assets': {
         'web.assets_backend': [

fusion_accounting_reports/data/cron.xml

@@ -0,0 +1,24 @@
+<?xml version="1.0" encoding="utf-8"?>
+<odoo noupdate="1">
+
+    <record id="cron_fusion_reports_anomaly_scan" model="ir.cron">
+        <field name="name">Fusion Reports - Daily Anomaly Scan</field>
+        <field name="model_id" ref="model_fusion_reports_cron"/>
+        <field name="state">code</field>
+        <field name="code">model._cron_anomaly_scan()</field>
+        <field name="interval_number">1</field>
+        <field name="interval_type">days</field>
+        <field name="active" eval="True"/>
+    </record>
+
+    <record id="cron_fusion_reports_mv_refresh" model="ir.cron">
+        <field name="name">Fusion Reports - MV Refresh</field>
+        <field name="model_id" ref="model_fusion_reports_cron"/>
+        <field name="state">code</field>
+        <field name="code">model._cron_mv_refresh()</field>
+        <field name="interval_number">15</field>
+        <field name="interval_type">minutes</field>
+        <field name="active" eval="True"/>
+    </record>
+
+</odoo>

fusion_accounting_reports/data/sql/create_mv_account_balance.sql

@@ -0,0 +1,31 @@
+-- Materialized view: per-account aggregated balances by year-month.
+-- Used by GL drill-down + trial balance for large DBs.
+-- Refresh strategy: cron every 15 minutes (Task 25); CONCURRENTLY-capable
+-- thanks to the unique index.
+
+CREATE MATERIALIZED VIEW IF NOT EXISTS fusion_account_balance_mv AS
+SELECT
+    ROW_NUMBER() OVER (
+        ORDER BY account_id, company_id, DATE_TRUNC('month', date)
+    )::INTEGER AS id,
+    account_id,
+    company_id,
+    DATE_TRUNC('month', date)::date AS period_month,
+    SUM(debit) AS debit,
+    SUM(credit) AS credit,
+    SUM(balance) AS balance,
+    COUNT(*) AS line_count
+FROM account_move_line
+WHERE parent_state = 'posted'
+GROUP BY account_id, company_id, DATE_TRUNC('month', date);
+
+-- The (account_id, company_id, period_month) tuple is the natural key.
+-- We mark it UNIQUE so REFRESH MATERIALIZED VIEW CONCURRENTLY is allowed.
+CREATE UNIQUE INDEX IF NOT EXISTS fusion_account_balance_mv_pkey
+    ON fusion_account_balance_mv (account_id, company_id, period_month);
+-- A separate index on the synthetic id is required by Odoo's ORM, which
+-- expects every model row to be addressable by `id`.
+CREATE UNIQUE INDEX IF NOT EXISTS fusion_account_balance_mv_id_idx
+    ON fusion_account_balance_mv (id);
+CREATE INDEX IF NOT EXISTS fusion_account_balance_mv_company_month
+    ON fusion_account_balance_mv (company_id, period_month);

fusion_accounting_reports/models/__init__.py

@@ -2,3 +2,5 @@ from . import fusion_report
 from . import fusion_report_engine
 from . import fusion_report_commentary
 from . import fusion_report_anomaly
+from . import fusion_account_balance_mv
+from . import fusion_reports_cron

fusion_accounting_reports/models/fusion_account_balance_mv.py

@@ -0,0 +1,80 @@
+"""Materialized view of per-account-per-month balances.
+
+Created lazily by init() (called by Odoo on install/upgrade). Refresh
+via the model's _refresh() method or via cron (Task 25)."""
+
+import logging
+import os
+
+from odoo import api, fields, models
+
+_logger = logging.getLogger(__name__)
+
+
+class FusionAccountBalanceMV(models.Model):
+    _name = "fusion.account.balance.mv"
+    _description = "MV of per-account per-month aggregated balances"
+    _auto = False
+    _table = "fusion_account_balance_mv"
+    _order = "period_month desc, account_id"
+
+    account_id = fields.Many2one('account.account', readonly=True)
+    company_id = fields.Many2one('res.company', readonly=True)
+    period_month = fields.Date(readonly=True)
+    debit = fields.Float(readonly=True)
+    credit = fields.Float(readonly=True)
+    balance = fields.Float(readonly=True)
+    line_count = fields.Integer(readonly=True)
+
+    def init(self):
+        # If the MV exists but is missing the synthetic `id` column (e.g. from
+        # an earlier dev install), drop it so the new schema applies cleanly.
+        self.env.cr.execute(
+            """
+            SELECT 1
+            FROM pg_matviews mv
+            JOIN pg_attribute a
+              ON a.attrelid = (mv.schemaname || '.' || mv.matviewname)::regclass
+             AND a.attname = 'id'
+            WHERE mv.matviewname = 'fusion_account_balance_mv'
+            """
+        )
+        if not self.env.cr.fetchone():
+            self.env.cr.execute(
+                "DROP MATERIALIZED VIEW IF EXISTS fusion_account_balance_mv"
+            )
+        sql_path = os.path.join(
+            os.path.dirname(__file__), '..', 'data', 'sql',
+            'create_mv_account_balance.sql',
+        )
+        with open(sql_path, 'r') as f:
+            self.env.cr.execute(f.read())
+        _logger.info(
+            "fusion_account_balance_mv: created/verified MV + indexes")
+
+    @api.model
+    def _refresh(self, *, concurrently=True):
+        """Refresh the MV. Falls back to non-concurrent if CONCURRENTLY fails.
+
+        REFRESH MATERIALIZED VIEW CONCURRENTLY requires the MV to be already
+        populated and an autocommit-capable cursor; the cron path in Task 25
+        opens a dedicated cursor for that. This helper keeps callers safe by
+        retrying without CONCURRENTLY on failure."""
+        keyword = "CONCURRENTLY" if concurrently else ""
+        try:
+            self.env.cr.execute(
+                f"REFRESH MATERIALIZED VIEW {keyword} fusion_account_balance_mv"
+            )
+            _logger.debug(
+                "fusion_account_balance_mv refreshed (%s)",
+                'concurrent' if concurrently else 'blocking',
+            )
+        except Exception as e:
+            if concurrently:
+                _logger.warning(
+                    "Concurrent MV refresh failed (%s); falling back", e)
+                self.env.cr.execute(
+                    "REFRESH MATERIALIZED VIEW fusion_account_balance_mv"
+                )
+            else:
+                raise

fusion_accounting_reports/models/fusion_reports_cron.py

@@ -0,0 +1,117 @@
+"""Cron handlers for fusion_accounting_reports.
+
+Two scheduled jobs:
+- _cron_anomaly_scan: daily P&L variance scan -> persist anomalies
+- _cron_mv_refresh: every 15 min CONCURRENTLY refresh the MV"""
+
+import logging
+from datetime import timedelta
+
+import odoo
+from odoo import api, fields, models
+
+from ..services.anomaly_detection import detect
+from ..services.date_periods import month_bounds
+
+_logger = logging.getLogger(__name__)
+
+
+class FusionReportsCron(models.AbstractModel):
+    _name = "fusion.reports.cron"
+    _description = "Fusion Reports Cron Handlers"
+
+    @api.model
+    def _cron_anomaly_scan(self):
+        """Run last-month P&L vs prior-year-same-month and persist anomalies."""
+        today = fields.Date.today()
+        # Walk back into the previous full calendar month.
+        last_month = today.replace(day=1) - timedelta(days=1)
+        period = month_bounds(last_month)
+
+        Report = self.env['fusion.report'].sudo()
+        Anomaly = self.env['fusion.report.anomaly'].sudo()
+        engine = self.env['fusion.report.engine']
+
+        for company in self.env['res.company'].search([]):
+            try:
+                pnl_def = Report.search(
+                    [
+                        ('report_type', '=', 'pnl'),
+                        '|', ('company_id', '=', company.id),
+                        ('company_id', '=', False),
+                    ],
+                    limit=1,
+                )
+                if not pnl_def:
+                    continue
+                result = engine.compute_pnl(
+                    period,
+                    comparison='previous_year',
+                    company_id=company.id,
+                )
+                anomalies = detect(result)
+                for a in anomalies:
+                    existing = Anomaly.search(
+                        [
+                            ('report_id', '=', pnl_def.id),
+                            ('company_id', '=', company.id),
+                            ('period_from', '=', period.date_from),
+                            ('period_to', '=', period.date_to),
+                            ('row_id', '=', a['row_id']),
+                        ],
+                        limit=1,
+                    )
+                    vals = {
+                        'report_id': pnl_def.id,
+                        'company_id': company.id,
+                        'period_from': period.date_from,
+                        'period_to': period.date_to,
+                        'row_id': a['row_id'],
+                        'label': a['label'],
+                        'current_amount': a['current_amount'],
+                        'comparison_amount': a['comparison_amount'],
+                        'variance_amount': a['variance_amount'],
+                        'variance_pct': a['variance_pct'],
+                        'severity': a['severity'],
+                        'direction': a['direction'],
+                    }
+                    if existing:
+                        existing.write(vals)
+                    else:
+                        Anomaly.create(vals)
+                _logger.info(
+                    "Anomaly scan for company %s: %d flagged",
+                    company.id, len(anomalies),
+                )
+            except Exception as e:
+                _logger.exception(
+                    "Anomaly scan failed for company %s: %s", company.id, e,
+                )
+
+    @api.model
+    def _cron_mv_refresh(self):
+        """REFRESH CONCURRENTLY via dedicated autocommit cursor.
+
+        REFRESH MATERIALIZED VIEW CONCURRENTLY cannot run inside a
+        transaction block, so we open a separate connection with autocommit
+        enabled. The blocking REFRESH is used as a fallback if the
+        concurrent path fails (e.g. on a cold MV with no rows yet)."""
+        try:
+            db_name = self.env.cr.dbname
+            db = odoo.sql_db.db_connect(db_name)
+            with db.cursor() as cron_cr:
+                cron_cr._cnx.set_session(autocommit=True)
+                cron_cr.execute(
+                    "REFRESH MATERIALIZED VIEW CONCURRENTLY "
+                    "fusion_account_balance_mv"
+                )
+            _logger.debug("MV refresh CONCURRENTLY succeeded")
+        except Exception as e:
+            _logger.warning(
+                "CONCURRENTLY refresh failed (%s); blocking fallback", e)
+            try:
+                self.env['fusion.account.balance.mv']._refresh(
+                    concurrently=False)
+            except Exception as e2:
+                _logger.exception(
+                    "Blocking MV refresh also failed: %s", e2)

fusion_accounting_reports/tests/__init__.py

@@ -13,3 +13,8 @@ from . import test_fusion_report_anomaly
 from . import test_reports_controller
 from . import test_reports_adapter
 from . import test_fusion_report_tools
+from . import test_engine_property
+from . import test_pnl_integration
+from . import test_bs_tb_integration
+from . import test_account_balance_mv
+from . import test_cron

fusion_accounting_reports/tests/test_account_balance_mv.py

@@ -0,0 +1,20 @@
+"""Tests for fusion_account_balance MV."""
+
+from odoo.tests.common import TransactionCase, tagged
+
+
+@tagged('post_install', '-at_install')
+class TestAccountBalanceMV(TransactionCase):
+
+    def test_mv_exists_and_is_queryable(self):
+        # Force initial refresh, then make sure the model can read it.
+        self.env['fusion.account.balance.mv']._refresh(concurrently=False)
+        rows = self.env['fusion.account.balance.mv'].search([], limit=5)
+        self.assertIsNotNone(rows)
+
+    def test_mv_refresh_concurrent(self):
+        # Try concurrent refresh; should either succeed or fall back gracefully.
+        try:
+            self.env['fusion.account.balance.mv']._refresh(concurrently=True)
+        except Exception as e:
+            self.fail(f"MV refresh raised: {e}")

fusion_accounting_reports/tests/test_bs_tb_integration.py

@@ -0,0 +1,54 @@
+"""Integration tests for balance sheet + trial balance."""
+
+from datetime import date
+
+from odoo.tests.common import TransactionCase, tagged
+
+from odoo.addons.fusion_accounting_reports.services.date_periods import Period
+
+
+@tagged('post_install', '-at_install', 'integration')
+class TestBalanceSheetIntegration(TransactionCase):
+
+    def test_balance_sheet_includes_total_assets(self):
+        result = self.env['fusion.report.engine'].compute_balance_sheet(
+            date(2026, 12, 31), company_id=self.env.company.id)
+        labels = [r['label'] for r in result['rows']]
+        self.assertIn('TOTAL ASSETS', labels)
+        self.assertIn('TOTAL LIABILITIES', labels)
+        self.assertIn('TOTAL EQUITY', labels)
+
+    def test_balance_sheet_total_assets_is_subtotal(self):
+        result = self.env['fusion.report.engine'].compute_balance_sheet(
+            date(2026, 12, 31), company_id=self.env.company.id)
+        ta = next(
+            (r for r in result['rows'] if r['label'] == 'TOTAL ASSETS'),
+            None,
+        )
+        self.assertIsNotNone(ta)
+        self.assertTrue(ta['is_subtotal'])
+
+    def test_balance_sheet_returns_period(self):
+        result = self.env['fusion.report.engine'].compute_balance_sheet(
+            date(2026, 4, 19), company_id=self.env.company.id)
+        self.assertEqual(result['period']['date_to'], '2026-04-19')
+
+
+@tagged('post_install', '-at_install', 'integration')
+class TestTrialBalanceIntegration(TransactionCase):
+
+    def test_trial_balance_returns_all_5_groups(self):
+        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)
+        labels = [r['label'] for r in result['rows']]
+        for label in ('Assets', 'Liabilities', 'Equity', 'Income', 'Expenses'):
+            self.assertIn(label, labels)
+
+    def test_trial_balance_has_total_subtotal(self):
+        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)
+        last = result['rows'][-1]
+        self.assertEqual(last['label'], 'Total (should be 0)')
+        self.assertTrue(last['is_subtotal'])

fusion_accounting_reports/tests/test_cron.py

@@ -0,0 +1,20 @@
+"""Tests for cron handlers."""
+
+from odoo.tests.common import TransactionCase, tagged
+
+
+@tagged('post_install', '-at_install')
+class TestFusionReportsCron(TransactionCase):
+
+    def setUp(self):
+        super().setUp()
+        self.cron = self.env['fusion.reports.cron']
+
+    def test_cron_mv_refresh_does_not_raise(self):
+        # Smoke test: the cron must complete without raising even if the
+        # CONCURRENTLY path fails on a cold MV (the handler falls back).
+        self.cron._cron_mv_refresh()
+
+    def test_cron_anomaly_scan_does_not_raise(self):
+        # Smoke test: scan all companies, persist anomalies, no exceptions.
+        self.cron._cron_anomaly_scan()

fusion_accounting_reports/tests/test_engine_property.py

@@ -0,0 +1,156 @@
+"""Property-based invariant tests for the reports engine.
+
+Hypothesis generates random scenarios; we assert mathematical invariants
+that must hold regardless of input."""
+
+from datetime import date, timedelta
+
+from hypothesis import HealthCheck, given, settings, strategies as st
+
+from odoo.tests.common import TransactionCase, tagged
+
+from odoo.addons.fusion_accounting_reports.services.date_periods import (
+    Period,
+    comparison_period,
+    fiscal_year_bounds,
+    month_bounds,
+    quarter_bounds,
+)
+from odoo.addons.fusion_accounting_reports.services.line_resolver import resolve
+from odoo.addons.fusion_accounting_reports.services.totaling import (
+    TotalLine,
+    aggregate,
+    is_balanced,
+)
+
+
+@tagged('post_install', '-at_install', 'property_based')
+class TestServiceInvariants(TransactionCase):
+    """Pure-Python invariants - fast, no DB writes."""
+
+    @given(d=st.dates(min_value=date(2020, 1, 1), max_value=date(2030, 12, 31)))
+    @settings(max_examples=100, deadline=2000,
+              suppress_health_check=[HealthCheck.function_scoped_fixture])
+    def test_fiscal_year_contains_reference_date(self, d):
+        period = fiscal_year_bounds(d)
+        self.assertLessEqual(period.date_from, d)
+        self.assertGreaterEqual(period.date_to, d)
+
+    @given(d=st.dates(min_value=date(2020, 1, 1), max_value=date(2030, 12, 31)))
+    @settings(max_examples=50, deadline=2000,
+              suppress_health_check=[HealthCheck.function_scoped_fixture])
+    def test_month_bounds_first_to_last_day(self, d):
+        period = month_bounds(d)
+        self.assertEqual(period.date_from.day, 1)
+        # Last day of month: adding 1 day rolls into the next month
+        next_day = period.date_to + timedelta(days=1)
+        self.assertNotEqual(next_day.month, period.date_to.month)
+
+    @given(d=st.dates(min_value=date(2020, 1, 1), max_value=date(2030, 12, 31)))
+    @settings(max_examples=50, deadline=2000,
+              suppress_health_check=[HealthCheck.function_scoped_fixture])
+    def test_quarter_bounds_three_months(self, d):
+        period = quarter_bounds(d)
+        # Quarter starts on month 1, 4, 7, or 10 and is exactly 3 months
+        self.assertIn(period.date_from.month, (1, 4, 7, 10))
+        self.assertEqual(period.date_from.day, 1)
+        self.assertGreaterEqual(period.date_to, d)
+        self.assertLessEqual(period.date_from, d)
+
+    @given(
+        debits=st.lists(
+            st.floats(min_value=0, max_value=10000,
+                      allow_nan=False, allow_infinity=False),
+            min_size=1, max_size=20,
+        ),
+    )
+    @settings(max_examples=50, deadline=2000,
+              suppress_health_check=[HealthCheck.function_scoped_fixture])
+    def test_aggregate_sum_equals_input_sum(self, debits):
+        lines = [
+            {'debit': d, 'credit': 0, 'balance': d, 'account_id': 1}
+            for d in debits
+        ]
+        result = aggregate(lines)
+        self.assertAlmostEqual(result.debit, sum(debits), places=2)
+        self.assertEqual(result.line_count, len(lines))
+
+    @given(
+        amounts=st.lists(
+            st.floats(min_value=1.0, max_value=100000,
+                      allow_nan=False, allow_infinity=False),
+            min_size=4, max_size=10,
+        ),
+    )
+    @settings(max_examples=50, deadline=2000,
+              suppress_health_check=[HealthCheck.function_scoped_fixture])
+    def test_balanced_iff_debits_equal_credits(self, amounts):
+        # Build a perfectly balanced ledger: half debits, half credits scaled
+        # so the totals match exactly.
+        half = len(amounts) // 2
+        debits = amounts[:half]
+        credits = amounts[half:half * 2]
+        if not credits or sum(credits) == 0:
+            return
+        scale = sum(debits) / sum(credits)
+        scaled_credits = [c * scale for c in credits]
+        lines = [{'debit': d, 'credit': 0, 'balance': d} for d in debits]
+        lines += [
+            {'debit': 0, 'credit': c, 'balance': -c} for c in scaled_credits
+        ]
+        # Allow a generous tolerance to account for float scaling drift on
+        # extreme inputs; the invariant we care about is still that balanced
+        # books read as balanced.
+        self.assertTrue(is_balanced(lines, tolerance=1.0))
+
+    @given(
+        period_from=st.dates(min_value=date(2021, 1, 1),
+                             max_value=date(2026, 1, 1)),
+    )
+    @settings(max_examples=30, deadline=2000,
+              suppress_health_check=[HealthCheck.function_scoped_fixture])
+    def test_comparison_previous_year_is_one_year_earlier(self, period_from):
+        # Build a 30-day period to keep things simple
+        period_to = period_from + timedelta(days=30)
+        period = Period(period_from, period_to, 'test')
+        comp = comparison_period(period, 'previous_year')
+        self.assertIsNotNone(comp)
+        self.assertEqual(comp.date_from.year, period.date_from.year - 1)
+        self.assertEqual(comp.date_to.year, period.date_to.year - 1)
+
+
+@tagged('post_install', '-at_install', 'property_based')
+class TestLineResolverInvariants(TransactionCase):
+    """Invariants on the line_resolver."""
+
+    @given(
+        n_accounts=st.integers(min_value=1, max_value=20),
+        balance=st.floats(min_value=-10000, max_value=10000,
+                          allow_nan=False, allow_infinity=False),
+    )
+    @settings(max_examples=30, deadline=2000,
+              suppress_health_check=[HealthCheck.function_scoped_fixture])
+    def test_subtotal_equals_sum_of_above_rows(self, n_accounts, balance):
+        accounts_by_id = {
+            i: {'code': f'{i:04d}', 'name': f'Acct {i}',
+                'account_type': 'asset_cash'}
+            for i in range(n_accounts)
+        }
+        account_totals = {
+            i: TotalLine(balance=balance) for i in range(n_accounts)
+        }
+        line_specs = [
+            {'label': f'Acct {i}', 'account_id': i, 'sign': 1}
+            for i in range(n_accounts)
+        ]
+        line_specs.append({
+            'label': 'Subtotal', 'compute': 'subtotal',
+            'above': n_accounts, 'sign': 1,
+        })
+
+        rows = resolve(line_specs, account_totals=account_totals,
+                       accounts_by_id=accounts_by_id)
+        subtotal = rows[-1]
+        non_subtotals = [r for r in rows[:-1] if not r.get('is_subtotal')]
+        expected = sum(r['amount'] for r in non_subtotals)
+        self.assertAlmostEqual(subtotal['amount'], expected, places=2)

fusion_accounting_reports/tests/test_pnl_integration.py

@@ -0,0 +1,107 @@
+"""Integration test: P&L produces correct totals against known fixtures.
+
+Creates a small set of known invoices/bills and verifies that compute_pnl
+returns the expected Revenue, Expenses, Net Income."""
+
+from datetime import date
+
+from odoo.tests.common import TransactionCase, tagged
+
+from odoo.addons.fusion_accounting_reports.services.date_periods import Period
+
+
+@tagged('post_install', '-at_install', 'integration')
+class TestPnlIntegration(TransactionCase):
+
+    def setUp(self):
+        super().setUp()
+        self.partner = self.env['res.partner'].create(
+            {'name': 'P&L Test Partner'})
+        self.income_account = self.env['account.account'].search(
+            [('account_type', '=', 'income'),
+             ('company_ids', 'in', self.env.company.id)],
+            limit=1,
+        )
+        # Make a service product and pin an income account so invoice lines
+        # always book to a known revenue account regardless of localisation.
+        self.product = self.env['product.product'].create({
+            'name': 'Fusion P&L Test Service',
+            'type': 'service',
+        })
+        if self.income_account:
+            self.product.property_account_income_id = self.income_account
+
+    def _create_invoice(self, amount, *, date_=None, move_type='out_invoice'):
+        line_vals = {
+            'product_id': self.product.id,
+            'name': 'Test',
+            'quantity': 1,
+            'price_unit': amount,
+            'tax_ids': [(6, 0, [])],
+        }
+        if self.income_account:
+            line_vals['account_id'] = self.income_account.id
+        invoice = self.env['account.move'].create({
+            'move_type': move_type,
+            'partner_id': self.partner.id,
+            'invoice_date': date_ or date(2026, 6, 15),
+            'invoice_line_ids': [(0, 0, line_vals)],
+        })
+        invoice.action_post()
+        # The engine reads parent_state via raw SQL; force a flush so the
+        # field is materialised in the DB before we aggregate.
+        self.env.flush_all()
+        return invoice
+
+    def test_pnl_includes_invoice_revenue(self):
+        period = Period(date(2026, 1, 1), date(2026, 12, 31), 'Test 2026')
+        baseline = self.env['fusion.report.engine'].compute_pnl(
+            period, company_id=self.env.company.id)
+        baseline_labels = [r.get('label') for r in baseline['rows']]
+        revenue_baseline = next(
+            (r['amount'] for r in baseline['rows']
+             if r.get('label') == 'Revenue'),
+            None,
+        )
+        self.assertIsNotNone(
+            revenue_baseline,
+            msg=f"Revenue row not found; got labels: {baseline_labels}",
+        )
+
+        self._create_invoice(1000)
+
+        result = self.env['fusion.report.engine'].compute_pnl(
+            period, company_id=self.env.company.id)
+        revenue_after = next(
+            (r['amount'] for r in result['rows']
+             if r.get('label') == 'Revenue'),
+            None,
+        )
+        self.assertIsNotNone(revenue_after)
+
+        delta = revenue_after - revenue_baseline
+        self.assertAlmostEqual(
+            delta, 1000, places=0,
+            msg=f"Expected Revenue +1000, got {delta:.2f}",
+        )
+
+    def test_pnl_with_comparison_returns_both_periods(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'))
+        for row in result['rows']:
+            if row.get('amount_comparison') is not None:
+                self.assertIsInstance(row['amount_comparison'], (int, float))
+                return
+        # No row had comparison amounts -- still acceptable for empty periods.
+
+    def test_pnl_net_income_is_subtotal(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)
+        last = result['rows'][-1]
+        self.assertTrue(last['is_subtotal'])
+        self.assertEqual(last['label'], 'Net Income')