dodil/erp-brick-gl

DODIL ERP brick: general ledger

tests/test_audit.py

"""GL-59: the hash chain, the FEC and SAF-T, the change log."""
import hashlib
from decimal import Decimal

from conftest import ACCOUNTANT, ACCOUNTANT2, AUDITOR, BOOKKEEPER, api, must, refused
from sqlalchemy import text
from world import account_id, books, close_through, invoice, line, post


def fr_books():
    books(company="FR01", currency="EUR")
    post("INV-F1", "issue", "2026-03-10", [line(purpose="receivable", debit="1200.00"), line(purpose="revenue", credit="1000.00"),
         line(account="2200", credit="200.00")], currency="EUR", company_code="FR01")
    post("COST-F1", "x", "2026-05-10", [line(account="6100", debit="400.00"), line(account="2100", credit="400.00")], source_module="inventory",
         currency="EUR", company_code="FR01")
    post("pay_F1", "receipt", "2026-04-10", [line(account="1200", debit="1200.00"), line(purpose="receivable", credit="1200.00")], source_module="payments",
         currency="EUR", company_code="FR01")


def test_the_chain_verifies_after_many_postings_and_breaks_at_the_changed_journal(engine):
    fr_books()
    chain = must(api("GET", "/audit/verify-chain", AUDITOR, {"company_code": "FR01"}))
    assert chain == {"company_code": "FR01", "journals": 3, "met": True}
    rows = must(api("GET", "/journals", AUDITOR, {"company_code": "FR01"}))["journals"]
    assert sorted(r["chain_seq"] for r in rows) == [1, 2, 3] and all(len(r["row_hash"]) == 64 for r in rows)
    victim = [r for r in rows if r["chain_seq"] == 2][0]
    with engine.begin() as c:
        c.execute(text("UPDATE gl_journal_line SET functional_debit = 399.00 WHERE journal_id = :j AND line_no = 1"), {"j": int(victim["journal_id"])})
    broken = must(api("GET", "/audit/verify-chain", AUDITOR, {"company_code": "FR01"}))
    assert broken["met"] is False and broken["first_broken"]["journal_id"] == victim["journal_id"] and broken["first_broken"]["chain_seq"] == 2
    assert api("GET", "/audit/verify-chain", BOOKKEEPER, {})[0] == 403


def test_changing_a_header_column_breaks_it_too_but_a_reversal_does_not(engine):
    fr_books()
    rows = must(api("GET", "/journals", AUDITOR, {"company_code": "FR01"}))["journals"]
    first = [r for r in rows if r["chain_seq"] == 1][0]
    must(api("POST", f"/journals/{first['journal_id']}/reverse-journal", ACCOUNTANT, {"reason": "x"}))          # status flips; the hash does not cover it
    assert must(api("GET", "/audit/verify-chain", AUDITOR, {"company_code": "FR01"}))["met"] is True
    with engine.begin() as c:
        c.execute(text("UPDATE gl_journal SET memo = 'edited' WHERE journal_id = :j"), {"j": int(first["journal_id"])})
    assert must(api("GET", "/audit/verify-chain", AUDITOR, {"company_code": "FR01"}))["first_broken"]["chain_seq"] == 1


def test_the_fec_has_eighteen_columns_balances_and_is_stable():
    fr_books()
    out = must(api("POST", "/audit/export-fec", AUDITOR, {"fiscal_year": "FY2026", "siren": "123456789", "company_code": "FR01"}))
    assert out["filename"] == "123456789FEC20261231.txt" and out["object_key"].startswith("raw/audit/FR01/FY2026/")
    body = must(api("GET", f"/audit/exports/{out['export_id']}/content", AUDITOR))["content"]
    lines = body.splitlines()
    assert lines[0].split("\t")[0] == "JournalCode" and len(lines[0].split("\t")) == 18 and all(len(ln.split("\t")) == 18 for ln in lines)
    assert hashlib.sha256(body.encode()).hexdigest() == out["sha256"]
    assert out["total_debit"] == out["total_credit"] == "2800.00"
    nums = {}
    for ln in lines[1:]:
        c = ln.split("\t")
        nums.setdefault(c[0], set()).add(int(c[2]))
    assert nums["SALES"] == {1} and nums["MISC"] == {1} and nums["CASH"] == {1}          # numbers without gaps per journal code
    again = must(api("POST", "/audit/export-fec", AUDITOR, {"fiscal_year": "FY2026", "siren": "123456789", "company_code": "FR01"}))
    assert again["sha256"] == out["sha256"] and again["object_key"] == out["object_key"] and again["export_id"] != out["export_id"]
    assert api("POST", "/audit/export-fec", AUDITOR, {"fiscal_year": "FY2026", "siren": "12345", "company_code": "FR01"})[0] == 422
    assert api("POST", "/audit/export-fec", BOOKKEEPER, {"fiscal_year": "FY2026", "siren": "123456789", "company_code": "FR01"})[0] == 403
    listed = must(api("GET", "/audit/exports", AUDITOR, {"company_code": "FR01"}))["exports"]
    assert len(listed) == 2 and listed[0]["produced_by"] == "auditor@firm.example"


def test_the_fec_uses_alternative_numbers_and_the_next_years_a_nouveaux_equal_the_carry_forward():
    fr_books()
    must(api("POST", "/accounts/alternative-number", ACCOUNTANT, {"company_code": "FR01", "account_id": str(account_id("4000")), "alt_code": "706000",
                                                                   "valid_from": "2026-01-01"}))
    out = must(api("POST", "/audit/export-fec", AUDITOR, {"fiscal_year": "FY2026", "siren": "123456789", "company_code": "FR01"}))
    body = must(api("GET", f"/audit/exports/{out['export_id']}/content", AUDITOR))["content"]
    assert "706000" in body and "\t4000\t" not in body
    close_through("2026-12", company="FR01")
    must(api("POST", "/fiscal-years/FY2026/close-year", ACCOUNTANT, {"company_code": "FR01"}))
    out27 = must(api("POST", "/audit/export-fec", AUDITOR, {"fiscal_year": "FY2027", "siren": "123456789", "company_code": "FR01"}))
    body27 = must(api("GET", f"/audit/exports/{out27['export_id']}/content", AUDITOR))["content"]
    an = [ln.split("\t") for ln in body27.splitlines()[1:] if ln.startswith("AN\t")]
    assert an and sum(Decimal(c[11].replace(",", ".")) for c in an) == sum(Decimal(c[12].replace(",", ".")) for c in an)
    opening = {c[4]: Decimal(c[11].replace(",", ".")) - Decimal(c[12].replace(",", ".")) for c in an}
    tb = must(api("GET", "/reports/trial-balance", ACCOUNTANT, {"company_code": "FR01", "as_at": "2027-01-31"}))
    assert opening == {r["code"]: Decimal(r["opening"]) for r in tb["rows"] if Decimal(r["opening"]) != 0}


def test_saft_is_stored_with_its_checksum():
    fr_books()
    out = must(api("POST", "/audit/export-saft", AUDITOR, {"fiscal_year": "FY2026", "company_code": "FR01"}))
    body = must(api("GET", f"/audit/exports/{out['export_id']}/content", AUDITOR))["content"]
    assert "<GeneralLedgerEntries>" in body and "<NumberOfEntries>3</NumberOfEntries>" in body and "<TotalDebit>2800.00</TotalDebit>" in body
    assert out["filename"] == "SAFT_FR01_FY2026.xml" and out["sha256"] == hashlib.sha256(body.encode()).hexdigest()


def test_master_data_changes_are_in_the_change_log_with_before_and_after():
    books()
    must(api("POST", "/accounts", ACCOUNTANT, {"chart_id": "MAIN", "code": "6100", "name": "Premises rent"}))
    log = must(api("GET", "/audit/change-log", AUDITOR, {"entity": "account", "natural_key": "MAIN:6100"}))["changes"]
    assert log[-1]["before"]["name"] == "Rent" and log[-1]["after"]["name"] == "Premises rent" and log[-1]["by"] == "accountant@firm.example"
    assert log[0]["before"] is None
    _ = (ACCOUNTANT2, invoice, refused)