dodil/erp-brick-consolidation

DODIL ERP brick: consolidation

tests/fake_odoo.py

"""An Odoo 16 Enterprise database with account_consolidation, behind the loader's port (seams.OdooReader). Rows are what the spec's reads name.
`extra` more subsidiaries and `journals_each` journals per company: the loader's statements must not grow with the number of companies."""
from datetime import date
from decimal import Decimal

from brick_consolidation.infrastructure.seams import OdooReader

D = Decimal
ACCOUNTS = [(11, "G-1100", "end", [101]), (12, "G-3000", "hist", [102]), (13, "G-4000", "avg", [103])]


class FakeOdoo(OdooReader):
    def __init__(self, extra=0, journals_each=1, db="ACME"):
        self.db, self.calls = db, []
        self.company_ids = [1, 2] + [10 + i for i in range(extra)]
        self.rate = {1: D("100"), 2: D("80")}
        self.journal_rows, self.line_rows = [], []
        n = 0
        for cid in self.company_ids:
            for k in range(journals_each):
                n += 1
                self.journal_rows.append({"id": 40 + n, "period_id": 21, "company_period_id": 30 + cid, "auto_generated": True, "name": f"J{n}"})
                self.line_rows += [{"id": 1000 + 2 * n, "journal_id": 40 + n, "account_id": 11, "amount": D("50.00") + k, "note": ""},
                                   {"id": 1001 + 2 * n, "journal_id": 40 + n, "account_id": 12, "amount": D("-50.00") - k, "note": ""}]

    def chart(self, chart_id):
        self.calls.append("chart")
        return {"id": 7, "name": "ACME group", "currency": "GBP", "company_ids": self.company_ids} if chart_id == 7 else None

    def accounts(self, chart_id):
        return [{"id": i, "code": c, "name": c, "currency_mode": m, "account_ids": a, "invert_sign": False} for i, c, m, a in ACCOUNTS]

    def periods(self, chart_id):
        return [{"id": 21, "date_analysis_begin": date(2026, 10, 1), "date_analysis_end": date(2026, 10, 31), "state": "closed"},
                {"id": 22, "date_analysis_begin": date(2026, 11, 1), "date_analysis_end": date(2026, 11, 30), "state": "draft"}]

    def company_periods(self, period_ids):
        return [{"id": 30 + c, "period_id": 21, "company_id": c, "rate_consolidation": self.rate.get(c, D("100"))} for c in self.company_ids]

    def journals(self, period_ids, after, limit):
        self.calls.append(("journals", after))
        return [j for j in self.journal_rows if j["id"] > after and j["period_id"] in period_ids][:limit]

    def journal_lines(self, journal_ids):
        self.calls.append(("lines", len(journal_ids)))
        return [ln for ln in self.line_rows if ln["journal_id"] in journal_ids]

    def aggregates(self, chart_id, period_ids):
        bal: dict = {}
        for ln in self.line_rows:
            bal[ln["account_id"]] = bal.get(ln["account_id"], D(0)) + ln["amount"]
        debit = sum((ln["amount"] for ln in self.line_rows if ln["amount"] > 0), D(0))
        return {"balance": {21: bal}, "structure": {21: {c: self.rate.get(c, D("100")) for c in self.company_ids}},
                "journals": {21: {"count": len(self.journal_rows), "total": debit}}}


def gl_signed(gl, odoo, db="ACME"):
    """gl's migration signed: its companies and accounts mapped, and the group chart GROUP carrying the consolidation accounts."""
    codes = {1: "UK01", 2: "DE01", **{10 + i: f"S{i:02d}" for i in range(len(odoo.company_ids) - 2)}}
    for c in codes.values():
        if c not in gl.companies_:
            gl.company(c, "EUR", fy_end=12)
    gl.migration_states[db] = {"status": "signed", "cutover_date": None}                    # v3: gl's GET /migration/status?db=
    gl.maps[(db, "res.company")] = {str(k): v for k, v in codes.items()}
    gl.maps[(db, "account.account")] = {str(x): f"L{x}" for _, _, _, ids in ACCOUNTS for x in ids}
    for _, code, _, _ in ACCOUNTS:
        gl.chart[code] = code