dodil/erp-brick-consolidation

DODIL ERP brick: consolidation

tests/test_v3.py

"""consolidation v3: the deferred tax rate order (CON-07), gl's migration signed before the Odoo load (CON-34), the drill-down to gl's lines
(CON-11), the business timezone read from gl, and the disposal at the closing rate (CON-16, in test_slice_c1)."""
import json
from decimal import Decimal

from conftest import api, as_role
from fake_odoo import FakeOdoo
from test_slice_c1 import at, year_world
from test_slice_d import DOORS, RUN, RUNS, SYSTEM, odoo_world, rows_of
from world import ACC, CTRL, P, ok, world

from brick_consolidation.infrastructure import clock, seams

D = Decimal
HOLDER = {"holder_company": "DE01", "seller_company": "UK01", "inventory_amount": "12000.00", "margin_pct": "25"}


def _to_eliminate(gl, **extra):
    year_world(gl)
    ok(api("POST", f"{at('2026-12')}/reopen-group-period", as_role("group_controller", email="second@group.example", sub="second.1"), {"reason": "rate"}))
    ok(api("POST", f"{at('2026-12')}/ic-inventory", ACC, {"holder_company": "DE01", "seller_company": "UK01", "none_stated": True}))
    for step in ("collect-group-data", "translate-currencies"):
        ok(api("POST", f"{at('2026-12')}/{step}", ACC, {}))
    ok(api("POST", f"{at('2026-12')}/ic-inventory", ACC, {**HOLDER, **extra}))
    return api("POST", f"{at('2026-12')}/run-eliminations", ACC, {})


def test_con_07_the_entry_rate_wins_then_gls_statutory_rate_on_the_period_end_then_a_refusal_never_zero(gl):
    status, body = _to_eliminate(gl)                                                   # no rate on the entry, none in gl
    assert status == 409 and "tax_rate_known" in str(body) and "DE01" in str(body) and "2026-12-31" in str(body)
    assert api("GET", f"{at('2026-12')}/readiness", CTRL)[1]["gates"][3]["met"] is False
    gl.tax_rates["DE01"] = [("2026-01-01", "0.25"), ("2027-01-01", "0.40")]            # the rate in force on the 31st, not the later one
    ok(api("POST", f"{at('2026-12')}/run-eliminations", ACC, {}))
    assert _deferred() == D("750.00")                                                   # 25% of the 3,000.00 stock profit, not 30% and not 40%


def test_con_07_a_rate_entered_on_the_entry_is_the_one_used(gl):
    gl.tax_rates["DE01"] = [("2026-01-01", "0.25")]
    ok(_to_eliminate(gl, tax_rate_pct="30"))
    assert _deferred() == D("900.00")


def _deferred():
    """The deferred tax asset of the latest elimination run of 2026-12 (an earlier run's entries are reversed by the later one)."""
    from world import db
    for run in (3, 2, 1):
        got = db(lambda r: r.data.entries("GRP", "2026-12", run))
        if got:
            return sum((ln["amount"] for e in got for ln in e["lines"] if ln["group_account_id"] == "G-1450"), D(0))


def test_con_34_the_odoo_load_proceeds_only_when_gls_migration_is_signed(gl):
    odoo_world(gl)
    for state in ("none", "loaded", "reconciled"):
        if state == "none":
            gl.migration_states.pop("ACME")
        else:
            gl.migration_states["ACME"] = {"status": state, "cutover_date": None}
        status, body = api("POST", RUNS, SYSTEM, {"cons_migration_run_id": RUN, "doors_json": DOORS})
        assert status == 409 and body["gate"] == "gl_signed" and state in body["reason"], (state, body)
        assert rows_of("cons_group") == [] and rows_of("cons_migration_run") == []     # nothing written
    gl.migration_states["ACME"] = {"status": "signed", "cutover_date": "2026-10-31"}
    status, body = api("POST", RUNS, SYSTEM, {"cons_migration_run_id": RUN, "doors_json": json.dumps({**json.loads(DOORS), "carried_periods": ["2026-10", "2026-11"]})})
    assert status == 409 and "cutover" in str(body)                                    # nothing after gl's cut-over comes from Odoo
    gl.migration_states["ACME"] = {"status": "signed", "cutover_date": None}
    ok(api("POST", RUNS, SYSTEM, {"cons_migration_run_id": RUN, "doors_json": DOORS}))


def test_con_11_the_drill_down_asks_gl_under_the_caller_and_stops_at_an_upload(gl):
    world(gl)
    gl.line_rows[("DE01", "1100")] = [{"line_id": "L1", "amount": "100.00"}]
    url = f"/groups/GRP/group-periods/{P}/lines"
    got = ok(api("GET", url, ACC, {"company_code": "DE01", "group_account_id": "1100"}))
    assert got["source"] == "gl" and got["lines"] == [{"line_id": "L1", "amount": "100.00"}]
    first = [c for c in gl.calls if c[0] == "lines"][-1]
    assert first[3] == "2026-04-01" or first[3] <= "2026-10-01" and first[4] == "2026-10-31"   # the fiscal year to date, to the period's last day
    gl.hidden = {"DE01"}                                                                 # gl's company scope: the caller cannot see DE01
    assert api("GET", url, ACC, {"company_code": "DE01", "group_account_id": "1100"})[0] == 404
    assert api("GET", url, ACC, {"company_code": "NOPE", "group_account_id": "1100"})[0] == 404
    assert api("GET", url, ACC, {"company_code": "DE01", "group_account_id": "G-1100", "figure": "x"})[0] in (409, 422)


def test_the_business_timezone_is_gls_for_the_top_company_read_per_request(gl):
    world(gl)
    gl.timezone = "Asia/Tokyo"
    seen = []
    real = clock.business_timezone

    def spy():
        seen.append(real())
        return seen[-1]
    clock.business_timezone = spy
    try:
        ok(api("GET", f"/groups/GRP", ACC, {}))
        assert real() == "Europe/London"                                                 # the request is over: nothing stored
    finally:
        clock.business_timezone = real
    assert ("company", "UK01") in [c[:2] for c in gl.calls]                              # asked gl for the top company