185 lines
7.6 KiB
Python
185 lines
7.6 KiB
Python
"""
|
||
End-to-end Excel export test on small synthetic data (no large files).
|
||
|
||
Generates a miniature Amazon transaction workbook (pivot 'Sheet1' + raw data sheet,
|
||
7 preamble rows, header row 8), runs the pipeline, exports the A/R Aging workbook, then
|
||
reopens it to verify structure, cross-sheet formulas, and subtotal values.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from datetime import date
|
||
|
||
import openpyxl
|
||
import pytest
|
||
|
||
from app.core.excel_export import export_workbook, FIELD_ORDER, PREAMBLE, HEADER_LABELS
|
||
from app.core.pipeline import process
|
||
|
||
TOTAL_IDX = FIELD_ORDER.index("total")
|
||
|
||
|
||
def _row(date_time, settlement_id, txn_type, account_type, total, order_id=""):
|
||
vals = [None] * len(FIELD_ORDER)
|
||
vals[FIELD_ORDER.index("date_time")] = date_time
|
||
vals[FIELD_ORDER.index("settlement_id")] = settlement_id
|
||
vals[FIELD_ORDER.index("txn_type")] = txn_type
|
||
vals[FIELD_ORDER.index("order_id")] = order_id
|
||
vals[FIELD_ORDER.index("account_type")] = account_type
|
||
vals[FIELD_ORDER.index("marketplace")] = "amazon.com"
|
||
vals[FIELD_ORDER.index("product_sales")] = total
|
||
vals[TOTAL_IDX] = total
|
||
return vals
|
||
|
||
|
||
def make_amazon_xlsx(path: str, order_rows: int = 12) -> None:
|
||
wb = openpyxl.Workbook()
|
||
pivot = wb.active
|
||
pivot.title = "Sheet1"
|
||
pivot["A1"] = "user pivot to ignore"
|
||
|
||
ws = wb.create_sheet("USA Amazon Transactions")
|
||
for line in PREAMBLE:
|
||
ws.append([line])
|
||
ws.append([HEADER_LABELS[f] for f in FIELD_ORDER]) # header row 8
|
||
|
||
# transfers (received boundary for 200; in-transit for 300)
|
||
ws.append(_row("Jan 6, 2026 9:00:00 AM PST", 200, "Transfer", "Standard Orders", -1000.0))
|
||
ws.append(_row("Jan 30, 2026 9:00:00 AM PST", 300, "Transfer", "Standard Orders", -2000.0))
|
||
ws.append(_row("Jan 12, 2026 9:00:00 AM PST", 250, "Transfer", "Invoiced Orders", -300.0))
|
||
|
||
# paid settlement 100 (Standard)
|
||
ws.append(_row("Jan 5, 2026 1:00:00 PM PST", 100, "Order", "Standard Orders", 1000.0, "o-100"))
|
||
# receivable settlement 200 (Standard): `order_rows` rows summing to 2000
|
||
per = 2000.0 / order_rows
|
||
for i in range(order_rows):
|
||
ws.append(_row("Jan 15, 2026 1:00:00 PM PST", 200, "Order", "Standard Orders", per, f"o200-{i}"))
|
||
# receivable settlement 300 (Standard): 500
|
||
ws.append(_row("Jan 31, 2026 1:00:00 PM PST", 300, "Order", "Standard Orders", 500.0, "o-300"))
|
||
# Invoiced: 150 paid, 250 receivable 80
|
||
ws.append(_row("Jan 10, 2026 1:00:00 PM PST", 150, "Order", "Invoiced Orders", 300.0, "i-150"))
|
||
ws.append(_row("Jan 20, 2026 1:00:00 PM PST", 250, "Order", "Invoiced Orders", 80.0, "i-250"))
|
||
wb.save(path)
|
||
|
||
|
||
@pytest.fixture()
|
||
def synth_file(tmp_path):
|
||
p = tmp_path / "USA synthetic.xlsx"
|
||
make_amazon_xlsx(str(p), order_rows=12)
|
||
return str(p)
|
||
|
||
|
||
_SUMMARY = {
|
||
"available": True, "marketplace": "USA", "currency": "USD", "reporting_month": "2026-01",
|
||
"opening_balance": 1000.0, "net_revenue": 2500.0, "disbursements": -500.0,
|
||
"closing_receivable": 3000.0,
|
||
"ledger": [
|
||
{"period": "Opening", "description": "Previous month closing balance",
|
||
"debit": None, "credit": None, "balance": 1000.0},
|
||
{"period": "Closing", "description": "Month-end receivable",
|
||
"debit": None, "credit": None, "balance": 3000.0},
|
||
],
|
||
}
|
||
_JOURNAL = {
|
||
"periods": [{"label": "01–31 Jan 2026"}],
|
||
"lines": [{"key": "Sales", "gl_account": "Sales:Amazon USA", "values": [2500.0], "total": 2500.0}],
|
||
"receivable": {"key": "Receivable", "gl_account": "AR:Amazon USA",
|
||
"values": [3000.0], "total": 3000.0},
|
||
}
|
||
|
||
|
||
def _run(synth_file, row_limit, with_reports=True):
|
||
result = process([synth_file], month_end=date(2026, 1, 31), clearing_lag_days=2)
|
||
out = str(synth_file) + ".aging.xlsx"
|
||
export_workbook(result, [synth_file], out,
|
||
reserves={("USA", "Standard Orders"): 0.0}, row_limit=row_limit,
|
||
summary=_SUMMARY if with_reports else None,
|
||
journal=_JOURNAL if with_reports else None)
|
||
return result, out
|
||
|
||
|
||
def test_export_structure_and_values(synth_file):
|
||
result, out = _run(synth_file, row_limit=1_048_576)
|
||
wb = openpyxl.load_workbook(out) # keep formulas
|
||
# sheet order
|
||
assert wb.sheetnames[:3] == ["Summary", "Detail", "COA"]
|
||
assert "USA" in wb.sheetnames
|
||
assert "Reconciliation" in wb.sheetnames
|
||
assert "Exceptions" in wb.sheetnames
|
||
assert "Processing Audit Trail" in wb.sheetnames
|
||
|
||
usa = wb["USA"]
|
||
# preamble + header
|
||
assert usa["A8"].value == "date/time"
|
||
assert usa["AD8"].value == "total"
|
||
# transfer rows at top have blank account type (excluded from SUMIFS)
|
||
assert usa["C9"].value == "Transfer"
|
||
assert usa["I9"].value in (None, "")
|
||
# subtotal formula references present
|
||
formulas = [c.value for col in usa.iter_cols(min_col=30, max_col=30)
|
||
for c in col if isinstance(c.value, str) and c.value.startswith("=SUMIFS")]
|
||
assert any("Standard Orders" in f for f in formulas)
|
||
assert any("Invoiced Orders" in f for f in formulas)
|
||
|
||
# Detail references marketplace subtotal cells & rounds
|
||
detail = wb["Detail"]
|
||
add_formula = detail["B5"].value
|
||
assert isinstance(add_formula, str) and "'USA'!" in add_formula
|
||
assert detail["B7"].value.startswith("=ROUND(")
|
||
|
||
# Summary references Detail USD cell
|
||
assert "Detail!" in wb["Summary"]["B6"].value
|
||
|
||
|
||
def test_receivable_value_computed_by_openpyxl(synth_file):
|
||
"""Load the generated tab, evaluate the SUMIFS ourselves, confirm 2500 + 80."""
|
||
result, out = _run(synth_file, row_limit=1_048_576)
|
||
wb = openpyxl.load_workbook(out)
|
||
usa = wb["USA"]
|
||
# sum the AD data cells by account type from column I
|
||
std = inv = 0.0
|
||
for r in range(9, usa.max_row + 1):
|
||
acct = usa.cell(row=r, column=FIELD_ORDER.index("account_type") + 1).value
|
||
tot = usa.cell(row=r, column=TOTAL_IDX + 1).value
|
||
if not isinstance(tot, (int, float)):
|
||
continue
|
||
if acct == "Standard Orders":
|
||
std += tot
|
||
elif acct == "Invoiced Orders":
|
||
inv += tot
|
||
assert std == pytest.approx(2500.0, abs=0.01)
|
||
assert inv == pytest.approx(80.0, abs=0.01)
|
||
# engine agrees
|
||
assert result.receivable.marketplaces["USA"].receivable_local == 2580
|
||
|
||
|
||
def test_full_workbook_includes_ledger_and_journal(synth_file):
|
||
"""The full audit workbook carries the AR Ledger and Journal Entry sheets."""
|
||
_, out = _run(synth_file, row_limit=1_048_576)
|
||
wb = openpyxl.load_workbook(out)
|
||
assert "AR Ledger" in wb.sheetnames
|
||
assert "Journal Entry" in wb.sheetnames
|
||
ledger = [c.value for row in wb["AR Ledger"].iter_rows() for c in row]
|
||
assert "Opening" in ledger and "Closing" in ledger and 3000.0 in ledger
|
||
journal = [c.value for row in wb["Journal Entry"].iter_rows() for c in row]
|
||
assert "Sales" in journal and "Receivable" in journal
|
||
|
||
|
||
def test_full_workbook_without_reports_still_valid(synth_file):
|
||
"""Ledger/journal sheets are optional — the workbook still builds without them."""
|
||
_, out = _run(synth_file, row_limit=1_048_576, with_reports=False)
|
||
wb = openpyxl.load_workbook(out)
|
||
assert "AR Ledger" not in wb.sheetnames
|
||
assert wb.sheetnames[:3] == ["Summary", "Detail", "COA"]
|
||
|
||
|
||
def test_sheet_split_when_over_row_limit(synth_file):
|
||
"""A tiny row_limit forces the USA marketplace across multiple sheets."""
|
||
result, out = _run(synth_file, row_limit=25)
|
||
wb = openpyxl.load_workbook(out)
|
||
usa_sheets = [n for n in wb.sheetnames if n.startswith("USA")]
|
||
assert len(usa_sheets) >= 2, wb.sheetnames
|
||
# Detail Additional-sales formula sums subtotal cells from every USA sheet
|
||
add = wb["Detail"]["B5"].value
|
||
for s in usa_sheets:
|
||
assert f"'{s}'!" in add
|