dodil/erp-brick-gl
DODIL ERP brick: general ledger
tests/fake_odoo.py
"""A fake Odoo: rows in memory, and Odoo's OWN aggregates computed from them (never from what gl loaded)."""
from datetime import date
from decimal import Decimal
from erp_platform import seam
from fastapi import FastAPI
from brick_gl.infrastructure import odoo
D = Decimal
def mk_line(id_, move_id, account_id, debit="0", credit="0", partner_id=None, tax_line_id=None, analytic=None, currency=None, amount_currency=None, name="line"):
return {"id": id_, "move_id": move_id, "account_id": account_id, "debit": D(debit), "credit": D(credit), "balance": D(debit) - D(credit),
"amount_currency": D(amount_currency) if amount_currency is not None else None, "currency": currency, "partner_id": partner_id, "name": name,
"tax_line_id": tax_line_id, "analytic_distribution": analytic, "display_type": "product"}
class FakeOdoo(odoo.OdooReader):
def __init__(self):
self.company = {"id": 1, "name": "Acme UK", "currency": "GBP", "fiscalyear_last_day": 31, "fiscalyear_last_month": 12, "period_lock_date": None,
"fiscalyear_lock_date": date(2026, 8, 31), "tax_lock_date": None, "hard_lock_date": None, "partner_id": 500}
self.acc = {1: dict(id=1, code="1100", name="Receivables", account_type="asset_receivable", currency=None),
2: dict(id=2, code="1200", name="Bank", account_type="asset_cash", currency=None),
3: dict(id=3, code="1210", name="Bank USD", account_type="asset_cash", currency="USD"),
4: dict(id=4, code="2100", name="Payables", account_type="liability_payable", currency=None),
5: dict(id=5, code="2200", name="VAT", account_type="liability_current", currency=None),
6: dict(id=6, code="4000", name="Revenue", account_type="income", currency=None),
7: dict(id=7, code="6000", name="Costs", account_type="expense", currency=None),
8: dict(id=8, code="3000", name="Current year earnings", account_type="equity_unaffected", currency=None)}
self.journal = {10: dict(id=10, code="INV", name="Sales", type="sale", default_account_id=None, currency=None),
11: dict(id=11, code="BNK", name="Bank", type="bank", default_account_id=2, currency=None),
12: dict(id=12, code="MISC", name="Misc", type="general", default_account_id=None, currency=None)}
self.mvs, self.mlines = {}, {}
self.analytic = [dict(id=51, name="Shop London", code="PC-LON", plan="Profit", company_id=1, active=True),
dict(id=52, name="Shop Warehouse", code="PC-WHS", plan="Profit", company_id=1, active=True),
dict(id=61, name="Dept IT", code="CC-IT", plan="Dept", company_id=1, active=True)]
self.rates_rows = [dict(id=1, date=date(2026, 3, 1), currency="USD", rate=D("1.25"), inverse_company_rate=D("0.8")),
dict(id=2, date=date(2026, 9, 1), currency="USD", rate=D("1.2658"), inverse_company_rate=None)]
self.recurring = []
self.deferrals = []
self.budget_rows = []
self.consolidation = []
self.repartition = [dict(tax_id=900, account_id=5, repartition_type="tax", document_type="invoice")]
self.defaults = {"property_account_receivable_id": 1, "property_account_payable_id": 4, "property_account_income_categ_id": 6}
self.drafts = 1
self.receivable_balance_override = None
self._build()
def add(self, id_, name, on, journal, lines, state="posted", reversed_entry_id=None, currency="GBP"):
self.mvs[id_] = {"id": id_, "name": name, "date": on, "state": state, "move_type": "entry", "journal_id": journal, "ref": name,
"reversed_entry_id": reversed_entry_id, "currency": currency, "company_id": 1}
self.mlines[id_] = lines
def _build(self):
self.add(100, "MISC/2024/0001", date(2024, 6, 1), 12, [mk_line(1000, 100, 2, "5000"), mk_line(1001, 100, 8, credit="5000")])
self.add(101, "INV/2025/0001", date(2025, 3, 10), 10, [mk_line(1010, 101, 1, "1200", partner_id=700), mk_line(1011, 101, 6, credit="1000", analytic={"51": 100}),
mk_line(1012, 101, 5, credit="200", tax_line_id=900)])
self.add(102, "BNK/2025/0001", date(2025, 4, 5), 11, [mk_line(1020, 102, 2, "1200"), mk_line(1021, 102, 1, credit="1200", partner_id=700)])
self.add(103, "MISC/2025/0001", date(2025, 11, 20), 12, [mk_line(1030, 103, 7, "300"), mk_line(1031, 103, 4, credit="300", partner_id=701)])
self.add(104, "INV/2026/0001", date(2026, 2, 10), 10, [mk_line(1040, 104, 1, "600", partner_id=700), mk_line(1041, 104, 6, credit="500", analytic={"51": 60, "52": 40}),
mk_line(1042, 104, 5, credit="100", tax_line_id=900)])
self.add(105, "BNK/2026/0001", date(2026, 3, 2), 11, [mk_line(1050, 105, 3, "800", currency="USD", amount_currency="1000"), mk_line(1051, 105, 6, credit="800")],
currency="USD")
self.add(106, "MISC/2026/0001", date(2026, 6, 15), 12, [mk_line(1060, 106, 4, "300", partner_id=701), mk_line(1061, 106, 7, credit="300")], reversed_entry_id=103)
self.add(107, "INV/2026/0002", date(2026, 9, 20), 10, [mk_line(1070, 107, 1, "120", partner_id=700), mk_line(1071, 107, 6, credit="100", analytic={"52": 100}),
mk_line(1072, 107, 5, credit="20", tax_line_id=900)])
self.add(108, "INV/2026/0003", date(2026, 10, 12), 10, [mk_line(1080, 108, 1, "240", partner_id=700), mk_line(1081, 108, 6, credit="200"),
mk_line(1082, 108, 5, credit="40", tax_line_id=900)])
self.add(109, "INV/2026/0004", date(2026, 9, 25), 10, [], state="draft")
# rows
def companies(self): return [self.company]
def accounts(self, company_id): return list(self.acc.values())
def journals(self, company_id): return list(self.journal.values())
def moves_(self, lo, hi):
return [m for m in self.mvs.values() if m["state"] == "posted" and (lo is None or m["date"] >= lo) and m["date"] <= hi]
def moves(self, company_id, date_from, date_to): return self.moves_(date_from, date_to)
def lines(self, move_ids): return [ln for mid in move_ids for ln in self.mlines.get(mid, [])]
def fiscal_years(self, company_id): return []
def rates(self, company_id, date_from, date_to): return [r for r in self.rates_rows if r["date"] <= date_to]
def analytic_accounts(self): return self.analytic
def recurring_moves(self, company_id, after): return self.recurring
def deferral_lines(self, company_id, after): return self.deferrals
def budgets(self, company_id): return self.budget_rows
def consolidation_accounts(self): return self.consolidation
def repartition_lines(self, tax_ids): return self.repartition
def company_defaults(self, company_id): return self.defaults
# Odoo's own aggregates
def _posted_lines(self, upto, lo=None):
for m in self.moves_(lo, upto):
for ln in self.mlines[m["id"]]:
yield m, ln
def opening_balances(self, company_id, before):
acc = {}
for m in self.mvs.values():
if m["state"] == "posted" and m["date"] < before:
for ln in self.mlines[m["id"]]:
acc[ln["account_id"]] = acc.get(ln["account_id"], D(0)) + ln["debit"] - ln["credit"]
return [{"account_id": a, "balance": b, "currency": None, "amount_currency": None} for a, b in sorted(acc.items()) if b != 0]
def tb_by_account_pc(self, company_id, cutover):
out = {}
pcs = {51: "PC-LON", 52: "PC-WHS"}
for m, ln in self._posted_lines(cutover):
dist = {int(k): D(str(v)) for k, v in (ln["analytic_distribution"] or {}).items() if int(k) in pcs} if m["date"] >= self.start else {}
shares = [(pcs[k], v / 100) for k, v in sorted(dist.items())] or [("", D(1))]
for pc, share in shares: # Odoo splits a line across its analytic accounts in proportion
cur = out.setdefault((ln["account_id"], pc), [D(0), D(0)])
cur[0] += ln["debit"] * share
cur[1] += ln["credit"] * share
return [{"account_id": a, "profit_centre": pc, "debit": v[0], "credit": v[1]} for (a, pc), v in out.items()]
def tb_by_account_period(self, company_id, cutover):
out = {}
for m, ln in self._posted_lines(cutover, self.start):
k = (ln["account_id"], f"{m['date'].year}-{m['date'].month:02d}")
cur = out.setdefault(k, [D(0), D(0)])
cur[0] += ln["debit"]
cur[1] += ln["credit"]
return [{"account_id": a, "period": p, "debit": v[0], "credit": v[1]} for (a, p), v in sorted(out.items(), key=lambda kv: kv[0][1])]
def partner_balances(self, company_id, cutover): return []
def foreign_balances(self, company_id, cutover):
out = {}
for m, ln in self._posted_lines(cutover):
if ln["currency"]:
k = (ln["account_id"], ln["currency"])
out[k] = out.get(k, D(0)) + ln["amount_currency"]
return [{"account_id": a, "currency": c, "amount_currency": v} for (a, c), v in out.items()]
def move_counts(self, company_id, cutover):
out = {}
for m in self.moves_(self.start, cutover):
k = (m["journal_id"], f"{m['date'].year}-{m['date'].month:02d}")
out[k] = out.get(k, 0) + 1
return [{"journal_id": j, "period": p, "count": n} for (j, p), n in out.items()]
def tax_by_quarter(self, company_id, cutover):
out = {}
for m, ln in self._posted_lines(cutover):
if ln["tax_line_id"] and m["date"] >= self.start:
k = (ln["tax_line_id"], f"{m['date'].year}-Q{(m['date'].month - 1) // 3 + 1}")
out[k] = out.get(k, D(0)) + ln["debit"] - ln["credit"]
return [{"tax_line_id": t, "quarter": q, "balance": v} for (t, q), v in out.items()]
def account_counts(self, company_id):
return [{**a, "count": 1} for a in self.acc.values()]
def rate_counts(self, company_id, cutover):
out = {}
for r in self.rates_rows:
if r["date"] <= cutover:
out[(r["currency"], r["date"].year)] = out.get((r["currency"], r["date"].year), 0) + 1
return [{"currency": c, "year": y, "count": n} for (c, y), n in out.items()]
def receivable_balance(self, company_id, cutover):
if self.receivable_balance_override is not None:
return self.receivable_balance_override
return sum((ln["debit"] - ln["credit"] for m, ln in self._posted_lines(cutover) if ln["account_id"] == 1), D(0))
def draft_move_count(self, company_id, cutover):
return sum(1 for m in self.mvs.values() if m["state"] == "draft" and m["date"] <= cutover)
start = date(2025, 1, 1)
def install(odoo_fake=None, partners=None, taxes=None):
"""Wire the fake reader and the business-partner / tax maps (their GET /migration-map answers)."""
f = odoo_fake or FakeOdoo()
odoo.ports.reader = f
bp, tax = FastAPI(), FastAPI()
partners = {500: 5000, 700: 7000, 701: 7001} if partners is None else partners
taxes = {900: "S"} if taxes is None else taxes
@bp.get("/migration-map/{db}/{model}")
def bp_map(db: str, model: str, ids: str):
return [{"odoo_model": model, "odoo_id": i, "ours": str(partners[int(i)])} for i in ids.split(",") if i and int(i) in partners]
@tax.get("/migration-map/{db}/{model}")
def tax_map(db: str, model: str, ids: str):
return [{"odoo_model": model, "odoo_id": i, "ours": taxes[int(i)]} for i in ids.split(",") if i and int(i) in taxes]
seam.register_local("business-partner", bp)
seam.register_local("tax", tax)
return f
def uninstall():
odoo.ports.reader = odoo.OdooReader()