dodil/erp-brick-consolidation

DODIL ERP brick: consolidation

tests/real_gl.py

"""The REAL gl brick (pip install -e ../erp-brick-gl) mounted in this process on its own SQLite bucket, the way a tier-1 composition mounts it:
consolidation's Seam reaches it by `seam.register_local("gl", app)` and every call is checked against gl's signed spec by conftest's checked_seams.
Seeded only through gl's own routes (companies, charts, accounts, the group mapping, rates, postings, closing periods)."""
import base64
import json
import pathlib
import re
from contextlib import contextmanager

import anyio
import brick_gl
import httpx
import yaml
from brick_gl.app import app as gl_app
from brick_gl.application import use_cases as gl_uc
from brick_gl.infrastructure.tables import Repository as GlRepository
from brick_gl.infrastructure.tables import md as gl_md
from erp_platform import seam
from erp_platform.seam import derived_id
from sqlalchemy import create_engine

_ROLES = yaml.safe_load((pathlib.Path(brick_gl.__file__).parent / "roles.yaml").read_text())["roles"]


def gl_person(email, permissions):
    payload = base64.urlsafe_b64encode(json.dumps({"permissions": list(permissions)}).encode()).decode()
    return {"X-Dodil-User": json.dumps({"sub": email, "email": email, "app_roles": []}), "X-Dodil-User-JWT": f"h.{payload.rstrip('=')}.s"}


ACCOUNTANT = gl_person("accountant@firm.example", _ROLES["accountant"])
OWNER = gl_person("owner@firm.example", _ROLES["owner"])
CONSOLIDATION = gl_person("cons-client@svc.example", ["gl:group_data:extract", "gl:company:read", "gl:fx:read", "gl:period:read", "gl:report:read"])
CONSOLIDATION_CALLER = CONSOLIDATION
POSTER = gl_person("ar-client@svc.example", ["gl:journal:post", "gl:period:read"])


async def _call(method, path, headers, body=None):
    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=gl_app), base_url="http://gl") as c:
        return await (c.get(path, headers=headers, params=body) if method == "GET" else c.request(method, path, headers=headers, json=body))


def gl(method, path, headers, body=None):
    r = anyio.run(_call, method, path, headers, body)
    try:
        return r.status_code, r.json()
    except ValueError:
        return r.status_code, r.text


def must(result):
    status, body = result
    assert status == 200, (status, body)
    return body


@contextmanager
def mounted(tmp_path):
    """gl on its own bucket (a SQLite file), registered as the neighbour `gl` for the duration."""
    engine = create_engine(f"sqlite:///{tmp_path / 'gl.db'}", connect_args={"check_same_thread": False, "timeout": 30})
    gl_md.create_all(engine)
    before = gl_uc.repository

    @contextmanager
    def repo():
        with engine.begin() as conn:
            yield GlRepository(conn)
    gl_uc.repository = repo
    seam.register_local("gl", gl_app)
    try:
        yield engine
    finally:
        seam.unregister_local("gl")
        gl_uc.repository = before
        engine.dispose()


FYE = 12
MAIN = [("1100", "Receivables", "asset", "receivable"), ("1200", "Bank", "asset", "bank"), ("1210", "Bank USD", "asset", "bank"),
        ("1300", "Due from group", "asset", "other"), ("1400", "Prepaid", "asset", "other"), ("1600", "Investment in subsidiaries", "asset", "other"),
        ("2100", "Accruals", "liability", "other"), ("2200", "VAT", "liability", "tax"), ("2300", "Payables", "liability", "payable"),
        ("2310", "Due to group", "liability", "other"), ("3000", "Share capital", "equity", "retained_earnings"), ("4000", "Revenue", "income", "other"),
        ("4100", "Group revenue", "income", "other"), ("4900", "FX gains", "income", "other"), ("5100", "Group cost", "expense", "other"),
        ("6000", "Costs", "expense", "other"), ("6900", "FX losses", "expense", "other"), ("6990", "Rounding", "expense", "other")]
PURPOSES = {"receivable": "1100", "revenue": "4000", "bank": "1200", "retained_earnings": "3000", "fx_gain": "4900", "fx_loss": "6900",
            "rounding": "6990", "bad_debt": "6000", "deposit": "2100", "accrual_balance": "2100", "prepaid": "1400"}
TO_GROUP = {"1200": "G-1100", "1300": "G-1300", "1600": "G-1600", "2310": "G-2300", "3000": "G-3000", "4100": "G-4100", "5100": "G-5100"}


def account(code):
    return str(derived_id("account", code))


def group_code_type(code):
    return {"1": "asset", "2": "liability", "3": "equity", "4": "income", "5": "expense", "6": "expense"}[code[2]]


def seed_books(group_codes):
    """Chart MAIN for UK01 (GBP) and DE01 (EUR), both on the calendar year; the group chart GROUP and the dated account mapping; the year open."""
    must(gl("POST", "/charts", ACCOUNTANT, {"chart_id": "MAIN", "name": "MAIN", "kind": "operational"}))
    for code, name, typ, sub in MAIN:
        must(gl("POST", "/accounts", ACCOUNTANT, {"chart_id": "MAIN", "code": code, "name": name, "type": typ, "subtype": sub}))
    for co, ccy, country in (("UK01", "GBP", "GB"), ("DE01", "EUR", "DE")):
        must(gl("POST", "/companies", ACCOUNTANT, {"company_code": co, "name": co, "country": country, "functional_currency": ccy, "chart_id": "MAIN", "financial_year_end_month": FYE}))
    for purpose, code in PURPOSES.items():
        must(gl("POST", "/account-map", ACCOUNTANT, {"chart_id": "MAIN", "purpose": purpose, "account_id": account(code), "valid_from": "2025-01-01"}))
    for co in ("UK01", "DE01"):
        for fy in ("FY2025", "FY2026"):
            must(gl("POST", "/fiscal-years/open-year", ACCOUNTANT, {"fiscal_year": fy, "company_code": co}))
    must(gl("POST", "/charts", ACCOUNTANT, {"chart_id": "GROUP", "kind": "group"}))
    for code in group_codes:
        must(gl("POST", "/accounts", ACCOUNTANT, {"chart_id": "GROUP", "code": code, "name": code, "type": group_code_type(code)}))
    ids = {a["code"]: str(a["account_id"]) for a in must(gl("GET", "/charts/GROUP/accounts", ACCOUNTANT))}
    for ours, theirs in TO_GROUP.items():
        must(gl("POST", "/accounts/group-mapping", ACCOUNTANT, {"account_id": account(ours), "group_account_id": ids[theirs], "valid_from": "2025-01-01"}))
    return ids


def rate(frm, to, value, on, kind):
    must(gl("POST", "/fx-rates", ACCOUNTANT, {"from_currency": frm, "to_currency": to, "rate": value, "rate_date": on, "rate_type": kind}))


def post(doc, on, company, currency, lines):
    body = {"source_module": "ar", "source_document": doc, "event": "x", "journal_date": on, "currency": currency, "company_code": company,
            "journal_id": str(derived_id(doc, "x")), "lines": [{"account_id": account(a), **kw} for a, kw in lines]}
    return must(gl("POST", "/journal-entry/journals", POSTER, body))


def match_off():
    """gl's close gate ic_matched needs a match run (gl's own ar/pur neighbours); the group close here is not about that, so the owner switches it off."""
    must(gl("POST", "/controls", OWNER, {"key": "ic_match_required", "value": "off"}))


def close_through(period, company):
    """Every month from 2025-01 to `period`, and the year end of FY2025 (its balances carry forward into 2026 as the opening)."""
    y, m = 2025, 1
    while f"{y:04d}-{m:02d}" <= period:
        must(gl("POST", f"/periods/{y:04d}-{m:02d}/close-period", ACCOUNTANT, {"company_code": company}))
        if (y, m) == (2025, 12):
            must(gl("POST", "/fiscal-years/FY2025/close-year", ACCOUNTANT, {"company_code": company}))
        y, m = (y, m + 1) if m < 12 else (y + 1, 1)


def real_rules(rules, ids):
    """world.RULES with the group accounts as gl's own ids (what GET /charts/GROUP/accounts answers)."""
    return json.loads(re.sub(r"G-\d{4}", lambda m: ids[m.group()], json.dumps(rules)))