dodil/erp-brick-consolidation
DODIL ERP brick: consolidation
tests/fakes.py
"""gl as its signed spec says it answers (module.yaml seams `with: consolidation`) — and nothing more.
Each answer carries only the fields the spec lists: companies {company_code, name, country, functional_currency, chart_id, bp_id,
financial_year_end_month, ledgers}; periods/open {date, company_code, period_id, open, status, reason, fix}; group-reporting/balances
{company_code, ledger_id, period_id, fiscal_year, period_status, functional_currency, group_chart_id, digest, rows[], unmapped[]};
fx-rates {currency, to, type, on, rate_date, rate, source} or 404 {gate: fx_rate, reason, fix}; intercompany/reconciliation {match_id,
as_at, matched[], breaks[], run_at} or 404. Every request is checked against the spec by `spec_check` before it is answered.
"""
import hashlib
import json
from decimal import Decimal
from contract_checks import spec_check
from brick_consolidation.infrastructure import seams
from brick_consolidation.infrastructure.seams import GlPort, Missing
D = Decimal
class FakeGl(GlPort):
def __init__(self):
self.companies_ = {}
self.status = {} # (company, period_id) -> closed | open
self.figures = {} # (company, period_id) -> dict(rows, unmapped)
self.rates = {} # (currency, kind, key) -> Decimal; key = date iso (closing, spot) or YYYY-MM (average)
self.recs = {} # (company, partner) -> reconciliation dict
self.calls = [] # (method, what, args) in order
self.chart = {} # account_id -> code of the group chart
self.versions = {} # fsv_id -> dict(kind, nodes, assignments), published
self.maps = {} # (db, model) -> {odoo_id (str): our natural key}: what gl's Odoo migration carried
self.tax_rates = {} # company -> [(valid_from iso, "0.25")]: gl's statutory_tax_rate rows (a decimal fraction)
self.timezone = "Europe/London" # gl's business_timezone, the bucket's one value
self.migration_states = {} # db -> {status, cutover_date}: gl's GL-34 answer (absent: status none)
self.line_rows = {} # (company, group_account_id) -> [gl line dict]
# ── set-up helpers (the test's side, not gl's API) ────────────────────────────────────────────────────────
def company(self, code, currency, name=None, fy_end=12):
self.companies_[code] = {"company_code": code, "name": name or code, "country": "XX", "functional_currency": currency,
"chart_id": "LOCAL", "bp_id": 1, "financial_year_end_month": fy_end, "ledgers": ["IFRS"]}
def set_rows(self, company, period_id, rows, unmapped=(), status="closed"):
"""rows: (group_account_id, ytd, period_amount[, partner, movement_type, segment]) -> gl's row shape."""
out = []
for r in rows:
out.append({"group_account_id": r[0], "partner_company": r[3] if len(r) > 3 else "", "movement_type": r[4] if len(r) > 4 else "",
"segment": r[5] if len(r) > 5 else "", "ytd_amount": str(D(str(r[1]))), "period_amount": str(D(str(r[2])))})
self.figures[(company, period_id)] = {"rows": out, "unmapped": [dict(u) for u in unmapped]}
self.status[(company, period_id)] = status
# ── the port ──────────────────────────────────────────────────────────────────────────────────────────────
def companies(self):
spec_check("gl", "GET", "/companies", None)
self.calls.append(("companies",))
return list(self.companies_.values())
def period_status(self, company, day):
spec_check("gl", "GET", "/periods/open", seams.period_params(company, day))
self.calls.append(("period", company))
st = self.status.get((company, f"{day.year:04d}-{day.month:02d}"), "none")
return {"date": day.isoformat(), "company_code": company, "period_id": f"{day.year:04d}-{day.month:02d}", "open": st == "open",
"status": st, "reason": "", "fix": ""}
def balances(self, company, ledger_id, period_id):
spec_check("gl", "GET", "/group-reporting/balances", seams.balances_params(company, ledger_id, period_id))
self.calls.append(("balances", company))
if company not in self.companies_:
raise Missing("gl", "company not visible", 404)
if (company, period_id) not in self.figures:
raise Missing("gl", "ledger_assigned: no ledger for the company and period — assign it", 422)
f = self.figures[(company, period_id)]
digest = hashlib.sha256(json.dumps(sorted(f["rows"], key=lambda r: json.dumps(r, sort_keys=True)), sort_keys=True).encode()).hexdigest()
return {"company_code": company, "ledger_id": ledger_id, "period_id": period_id, "fiscal_year": int(period_id[:4]),
"period_status": self.status[(company, period_id)], "functional_currency": self.companies_[company]["functional_currency"],
"group_chart_id": "GROUP", "digest": digest, "rows": f["rows"], "unmapped": f["unmapped"]}
def fx_rate(self, currency, on, kind, to):
if currency == to:
return D("1.000000")
spec_check("gl", "GET", f"/fx-rates/{currency}", seams.fx_params(on, kind, to))
self.calls.append(("fx", currency, kind, on.isoformat()))
key = f"{on.year:04d}-{on.month:02d}" if kind == "average" else on.isoformat()
if (currency, kind, key) not in self.rates:
raise Missing("gl", f"fx_rate: no {kind} rate for {currency} on {on} — add it in gl", 404)
return self.rates[(currency, kind, key)]
def reconciliation(self, company, partner, as_at):
spec_check("gl", "GET", "/intercompany/reconciliation", seams.reconciliation_params(company, partner, as_at))
self.calls.append(("reconciliation", company, partner))
return self.recs.get((company, partner)) or self.recs.get((partner, company))
def chart_accounts(self, chart_id):
spec_check("gl", "GET", f"/charts/{chart_id}/accounts", None)
self.calls.append(("chart", chart_id))
return [{"account_id": a, "code": c, "name": a, "type": "x", "subtype": ""} for a, c in sorted(self.chart.items())]
def statement_version(self, fsv_id, version_no=None):
spec_check("gl", "GET", f"/statement-versions/{fsv_id}", seams.version_params(version_no))
self.calls.append(("fsv", fsv_id))
if fsv_id not in self.versions:
raise Missing("gl", f"statement version {fsv_id} is not published", 404)
v = self.versions[fsv_id]
return {"fsv_id": fsv_id, "version_no": v.get("version_no", 1), "kind": v["kind"], "chart_id": "GROUP", "status": "published",
"nodes": v["nodes"], "assignments": v["assignments"]}
def company_row(self, company_code, on=None):
# gl's GL-61 row for consolidation lists `{ field: on, ... }` unquoted: YAML 1.1 reads `on` as the boolean True, so the spec's request_schema
# names a field "True" and the checker rejects the (specified) `on` query. Checked without it until erp-method quotes the key; reported.
spec_check("gl", "GET", f"/companies/{company_code}", None)
self.calls.append(("company", company_code, on.isoformat() if on else None))
if company_code not in self.companies_:
raise Missing("gl", "no such company", 404)
rows = [r for r in sorted(self.tax_rates.get(company_code, [])) if on is None or r[0] <= on.isoformat()]
return {**self.companies_[company_code], "statutory_tax_rate": rows[-1][1] if rows else None, "business_timezone": self.timezone}
def migration_status(self, db):
spec_check("gl", "GET", "/migration/status", seams.status_params(db))
self.calls.append(("migration_status", db))
got = self.migration_states.get(db)
return {"db": db, "run_id": f"odoo:{db}" if got else None, "status": got["status"] if got else "none",
"cutover_date": got.get("cutover_date") if got else None, "reconciled_at": None, "signed_by": None, "signed_at": None}
def account_lines(self, group_account_id, company, ledger_id, frm, to, limit, offset, caller=None):
spec_check("gl", "GET", f"/accounts/{group_account_id}/lines", seams.lines_params(company, ledger_id, frm, to, group_account_id, limit, offset))
self.calls.append(("lines", company, str(group_account_id), frm.isoformat(), to.isoformat()))
if company not in self.companies_ or company in getattr(self, "hidden", ()):
raise Missing("gl", "company is not visible", 404)
rows = self.line_rows.get((company, str(group_account_id)), [])
return {"account_id": str(group_account_id), "company_code": company, "ledger_id": ledger_id or "IFRS", "from": frm.isoformat(), "to": to.isoformat(),
"opening_balance": "0.00", "lines": rows[offset:offset + limit], "closing_balance": "0.00", "more": offset + limit < len(rows)}
def migration_map(self, db, model, ids):
out: dict = {}
wanted = sorted({str(i) for i in ids})
for n in range(0, len(wanted), seams.MAP_CHUNK):
chunk = wanted[n:n + seams.MAP_CHUNK]
spec_check("gl", "GET", f"/migration-map/{db}/{model}", seams.map_params(chunk))
self.calls.append(("map", model, len(chunk)))
held = self.maps.get((db, model), {})
out.update({i: held[i] for i in chunk if i in held}) # an id not migrated is omitted
return out
class FakeStore(seams.FilePort):
"""Where uploaded files go in tests: key -> (bytes, sha256)."""
def __init__(self):
self.put_ = {}
def put(self, key, data, sha256):
self.put_[key] = (data, sha256)