"""Books ledger list service (services/books.py; P4). SQLite fixtures per house style (mirrors tests/money/test_settlement.py): a PlusROI tenant with a tagged April + May and untagged strays, a second tenant to prove isolation, and SimpleNamespace actors for the authz cells. """ from datetime import date from decimal import Decimal from types import SimpleNamespace import pytest from sqlalchemy import create_engine from sqlalchemy.orm import Session from sqlalchemy.pool import StaticPool from app.models import Base, Expense, Invoice, Party, Project, Tenant from app.services import books from app.services.errors import ServiceError APRIL = date(2026, 4, 1) MAY = date(2026, 5, 1) @pytest.fixture() def s(): engine = create_engine( "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool ) Base.metadata.create_all(engine) with Session(engine) as session: yield session engine.dispose() @pytest.fixture() def books_fx(s): plusroi = Party(kind="org", name="PlusROI", slug="plusroi") bowden = Party(kind="org", name="Bowden Works", slug="bowden-works") copernic = Party(kind="org", name="Copernic", slug="copernic") acme = Party(kind="org", name="Acme 50%", slug="acme") # literal % in the name s.add_all([plusroi, bowden, copernic, acme]) s.flush() tenant = Tenant( party_id=plusroi.id, settings={"parallel_run": True, "partner_split": 0.5}, ) s.add(tenant) s.flush() proj = Project( tenant_id=tenant.id, operator_party_id=plusroi.id, client_party_id=copernic.id, name="Website revamp", ) s.add(proj) s.flush() def invoice( name, total, *, client, fee=None, month=APRIL, status="paid", external_id=None, currency="CAD", original_amount=None, original_currency=None, ): s.add( Invoice( tenant_id=tenant.id, name=name, status=status, currency=currency, manual_total=Decimal(total), fee=Decimal(fee) if fee else None, settlement_month=month, from_party_id=plusroi.id, to_party_id=client.id, external_id=external_id, invoice_date=month, original_amount=Decimal(original_amount) if original_amount else None, original_currency=original_currency, ) ) def expense(amount, category, *, paid_by=None, month=APRIL, project_id=None, vendor=None, external_invoice_id=None): s.add( Expense( tenant_id=tenant.id, amount=Decimal(amount), currency="CAD", category=category, paid_by_party_id=paid_by, settlement_month=month, project_id=project_id, vendor_party_id=vendor, external_invoice_id=external_invoice_id, source="sheet", ) ) # April receivables. invoice("A-1", "30000.00", client=copernic, fee="400.00", external_id="2026035") invoice("A-2 Acme 50%", "20000.00", client=acme, fee="247.88", external_id="2026036") invoice( "A-3 USD job", "5492.20", client=copernic, external_id="2026037", currency="CAD", original_amount="4000.00", original_currency="USD", ) # May receivables. invoice("M-1", "1000.00", client=copernic, fee="10.00", month=MAY, external_id="2026050") # Untagged receivable (the unsettled bucket). invoice("U-1", "999.99", client=copernic, month=None, status="pending_approval") # April payables. expense("18026.09", "Labour", paid_by=plusroi.id, project_id=proj.id, vendor=bowden.id) expense("2591.09", "Labour", paid_by=bowden.id, vendor=bowden.id) expense("644.14", "Client Advertising Spend", paid_by=bowden.id, vendor=copernic.id) # May payable. expense("100.00", "Labour", paid_by=plusroi.id, month=MAY) # Untagged payable. expense("55.55", "Other Expenses", paid_by=plusroi.id, month=None) s.flush() return SimpleNamespace( tenant=tenant, plusroi=plusroi, bowden=bowden, copernic=copernic, proj=proj ) def owner(tenant_id): return SimpleNamespace(tenant_id=tenant_id, capabilities={"can_see_income": True}) def worker(tenant_id): return SimpleNamespace(tenant_id=tenant_id, capabilities={"can_see_income": False}) # --- authz ------------------------------------------------------------------ def test_invoices_owner_only(s, books_fx): with pytest.raises(ServiceError) as err: books.list_invoices(s, worker(books_fx.tenant.id)) assert err.value.code == "INCOME_FORBIDDEN" def test_expenses_owner_only(s, books_fx): with pytest.raises(ServiceError) as err: books.list_expenses(s, worker(books_fx.tenant.id)) assert err.value.code == "INCOME_FORBIDDEN" def test_meta_owner_only(s, books_fx): with pytest.raises(ServiceError) as err: books.books_meta(s, worker(books_fx.tenant.id)) assert err.value.code == "INCOME_FORBIDDEN" # --- meta ------------------------------------------------------------------- def test_meta_parallel_run(s, books_fx): assert books.books_meta(s, owner(books_fx.tenant.id)) == {"parallel_run": True} # --- invoices --------------------------------------------------------------- def test_invoices_all_months_and_sums(s, books_fx): res = books.list_invoices(s, owner(books_fx.tenant.id)) assert res["total_count"] == 5 # 30000 + 20000 + 5492.20 + 1000 + 999.99 assert res["sum_amount"] == "57492.19" # 400 + 247.88 + 10 assert res["sum_fees"] == "657.88" # newest settlement month first; untagged (May's 2026-05 then April), # unsettled sorts to the end under nulls_last. months = [r["settlement_month"] for r in res["rows"]] assert months[0] == "2026-05" assert months[-1] is None def test_invoice_row_shape(s, books_fx): res = books.list_invoices(s, owner(books_fx.tenant.id), month="2026-05") assert res["total_count"] == 1 row = res["rows"][0] assert row["external_id"] == "2026050" assert row["client"] == "Copernic" assert row["amount"] == "1000.00" # manual_total surfaced as "amount" assert row["fee"] == "10.00" assert row["settlement_month"] == "2026-05" assert row["currency"] == "CAD" def test_invoice_usd_face_retained(s, books_fx): res = books.list_invoices(s, owner(books_fx.tenant.id), q="USD job") assert res["total_count"] == 1 row = res["rows"][0] assert row["original_currency"] == "USD" assert row["original_amount"] == "4000.00" assert row["amount"] == "5492.20" # book value stays CAD def test_invoices_month_filter(s, books_fx): res = books.list_invoices(s, owner(books_fx.tenant.id), month="2026-04") assert res["total_count"] == 3 assert res["sum_amount"] == "55492.20" assert res["sum_fees"] == "647.88" def test_invoices_unsettled_filter(s, books_fx): res = books.list_invoices(s, owner(books_fx.tenant.id), month="unsettled") assert res["total_count"] == 1 assert res["rows"][0]["name"] == "U-1" assert res["rows"][0]["settlement_month"] is None def test_invoices_status_filter(s, books_fx): res = books.list_invoices(s, owner(books_fx.tenant.id), status="pending_approval") assert res["total_count"] == 1 assert res["rows"][0]["name"] == "U-1" def test_invoices_bad_month(s, books_fx): with pytest.raises(ServiceError) as err: books.list_invoices(s, owner(books_fx.tenant.id), month="2026-13-99") assert err.value.code == "BAD_MONTH" def test_invoices_q_escaping(s, books_fx): # "50%" must be matched LITERALLY — an unescaped % would wildcard-match # everything, so a literal-match hit proves the escape works. res = books.list_invoices(s, owner(books_fx.tenant.id), q="50%") names = {r["name"] for r in res["rows"]} assert names == {"A-2 Acme 50%"} def test_invoices_q_matches_client_and_external_id(s, books_fx): by_client = books.list_invoices(s, owner(books_fx.tenant.id), q="copernic") assert by_client["total_count"] == 4 # every Copernic invoice by_ext = books.list_invoices(s, owner(books_fx.tenant.id), q="2026035") assert by_ext["total_count"] == 1 assert by_ext["rows"][0]["external_id"] == "2026035" def test_invoices_pagination_sums_over_full_set(s, books_fx): p1 = books.list_invoices(s, owner(books_fx.tenant.id), page=1, page_size=2) assert len(p1["rows"]) == 2 assert p1["total_count"] == 5 # the sum is the FULL filtered set, not just the page assert p1["sum_amount"] == "57492.19" p3 = books.list_invoices(s, owner(books_fx.tenant.id), page=3, page_size=2) assert len(p3["rows"]) == 1 # 5 rows over pages of 2 assert p3["sum_amount"] == "57492.19" def test_page_size_capped(s, books_fx): res = books.list_invoices(s, owner(books_fx.tenant.id), page_size=99999) assert res["page_size"] == books.MAX_PAGE_SIZE # --- expenses --------------------------------------------------------------- def test_expenses_all_and_sum(s, books_fx): res = books.list_expenses(s, owner(books_fx.tenant.id)) assert res["total_count"] == 5 # 18026.09 + 2591.09 + 644.14 + 100 + 55.55 assert res["sum_amount"] == "21416.87" def test_expense_row_shape_and_reimburse_flag(s, books_fx): res = books.list_expenses(s, owner(books_fx.tenant.id), month="2026-04") by_amount = {r["amount"]: r for r in res["rows"]} tenant_paid = by_amount["18026.09"] assert tenant_paid["paid_by"] == "PlusROI" assert tenant_paid["paid_by_is_tenant"] is True # muted, no reimbursement assert tenant_paid["project"] == "Website revamp" assert tenant_paid["vendor"] == "Bowden Works" bw_paid = by_amount["2591.09"] assert bw_paid["paid_by"] == "Bowden Works" assert bw_paid["paid_by_is_tenant"] is False # accented — reimbursable def test_expenses_category_filter_and_options(s, books_fx): res = books.list_expenses(s, owner(books_fx.tenant.id), category="Labour") assert res["total_count"] == 3 assert res["categories"] == ["Client Advertising Spend", "Labour", "Other Expenses"] assert res["paid_by_options"] == ["Bowden Works", "PlusROI"] def test_expenses_paid_by_filter(s, books_fx): res = books.list_expenses(s, owner(books_fx.tenant.id), paid_by="Bowden Works") assert res["total_count"] == 2 assert all(r["paid_by"] == "Bowden Works" for r in res["rows"]) def test_expenses_unsettled_filter(s, books_fx): res = books.list_expenses(s, owner(books_fx.tenant.id), month="unsettled") assert res["total_count"] == 1 assert res["rows"][0]["settlement_month"] is None assert res["rows"][0]["category"] == "Other Expenses" def test_expenses_q_matches_project_and_external(s, books_fx): res = books.list_expenses(s, owner(books_fx.tenant.id), q="revamp") assert res["total_count"] == 1 assert res["rows"][0]["project"] == "Website revamp" # --- tenant isolation ------------------------------------------------------- @pytest.fixture() def other_tenant(s): other_party = Party(kind="org", name="OtherCo", slug="otherco") s.add(other_party) s.flush() other = Tenant(party_id=other_party.id, settings={"parallel_run": False}) s.add(other) s.flush() s.add( Invoice( tenant_id=other.id, name="X-1", status="paid", currency="CAD", manual_total=Decimal("77777.77"), settlement_month=APRIL, ) ) s.add( Expense( tenant_id=other.id, amount=Decimal("88888.88"), currency="CAD", category="Secret", settlement_month=APRIL, source="sheet", ) ) s.flush() return other def test_invoice_tenant_isolation(s, books_fx, other_tenant): res = books.list_invoices(s, owner(books_fx.tenant.id)) assert res["total_count"] == 5 # the other tenant's X-1 is invisible assert "77777.77" not in {r["amount"] for r in res["rows"]} # and the reverse: the other tenant sees only its own row other = books.list_invoices(s, owner(other_tenant.id)) assert other["total_count"] == 1 assert other["sum_amount"] == "77777.77" def test_expense_tenant_isolation(s, books_fx, other_tenant): res = books.list_expenses(s, owner(books_fx.tenant.id)) assert res["total_count"] == 5 assert "Secret" not in res["categories"] def test_meta_reads_own_tenant(s, books_fx, other_tenant): assert books.books_meta(s, owner(other_tenant.id)) == {"parallel_run": False} # --- empty states ----------------------------------------------------------- def test_empty_invoices_shape(s, other_tenant): # a tenant with a filter matching nothing res = books.list_invoices(s, owner(other_tenant.id), month="2019-01") assert res["rows"] == [] assert res["total_count"] == 0 assert res["sum_amount"] == "0.00" assert res["sum_fees"] == "0.00" def test_empty_expenses_shape(s, other_tenant): res = books.list_expenses(s, owner(other_tenant.id), month="2019-01") assert res["rows"] == [] assert res["total_count"] == 0 assert res["sum_amount"] == "0.00" # options are computed over the whole tenant, independent of the filter assert res["categories"] == ["Secret"]