feat(fusion_accounting_reports): drill_down_resolver service

Pure-Python helper that, given an account_id and a date range, fetches
posted account.move.line records and returns a flat list of dicts ready
for the drill-down OWL dialog. Used by the engine's drill_down() method.

3 new tests, 35 total passing.

Made-with: Cursor

fusion_accounting_reports/__manifest__.py

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

fusion_accounting_reports/services/__init__.py

@@ -3,3 +3,4 @@ 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/tests/__init__.py

@@ -2,3 +2,4 @@ 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

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)