aiolabs/libra
8
0
Fork 1
Code Issues 36 Pull requests Projects Releases 6 Packages Wiki Activity Actions

Compare commits

base: aiolabs:116df46d389306421a3113f1b991fbde5a9f409d
Branches Tags
aiolabs:main
aiolabs:fix/authorization-security-refactor
aiolabs:fix/concurrency-protection
aiolabs:v0.4.0
aiolabs:v0.3.0
aiolabs:v0.2.2
aiolabs:v0.2.1
aiolabs:v0.2.0
aiolabs:v0.1.0
aiolabs:v0.0.9
aiolabs:v0.0.8
aiolabs:v0.0.7
aiolabs:v0.0.6
aiolabs:v0.0.5
aiolabs:v0.0.4
aiolabs:v0.0.3
aiolabs:v0.0.2
aiolabs:v0.0.1
..
compare: aiolabs:9c88993c136c095311ff5ad4ebdb2acdda0bbe9d
Branches Tags
aiolabs:main
aiolabs:fix/authorization-security-refactor
aiolabs:fix/concurrency-protection
aiolabs:v0.4.0
aiolabs:v0.3.0
aiolabs:v0.2.2
aiolabs:v0.2.1
aiolabs:v0.2.0
aiolabs:v0.1.0
aiolabs:v0.0.9
aiolabs:v0.0.8
aiolabs:v0.0.7
aiolabs:v0.0.6
aiolabs:v0.0.5
aiolabs:v0.0.4
aiolabs:v0.0.3
aiolabs:v0.0.2
aiolabs:v0.0.1

No commits in common. "116df46d389306421a3113f1b991fbde5a9f409d" and "9c88993c136c095311ff5ad4ebdb2acdda0bbe9d" have entirely different histories.
116df46d38 ... 9c88993c13

19 changed files with 8 additions and 4306 deletions
Download patch file Download diff file Expand all files Collapse all files

133
beancount_format.py
View file

@ -804,139 +804,6 @@ def format_net_settlement_entry(
) )
def format_fiat_net_settlement_entry(
user_id: str,
cash_account: str,
receivable_account: str,
payable_account: Optional[str],
credit_account: Optional[str],
cash_paid_fiat: Decimal,
total_receivable_fiat: Decimal,
total_payable_fiat: Decimal,
credit_overflow_fiat: Decimal,
fiat_currency: str,
description: str,
entry_date: date,
payment_method: str = "cash",
reference: Optional[str] = None,
settled_entry_links: Optional[List[str]] = None,
) -> Dict[str, Any]:
"""Fiat cash settlement that nets receivable and payable for one user.
Implements the contract from libra-#33 (settlement netting) and
libra-#41 (credit-balance overflow). Builds a 2- to 4-leg transaction
depending on what the user has open:
- Cash + Receivable only (2-leg) — pure receivable, exact pay
- Cash + Receivable + Credit (3-leg) — overpay against pure receivable
- Cash + Receivable + Payable (3-leg) — nancy's #33 scenario, exact pay
- Cash + Receivable + Payable + Credit (4-leg) — net + overpay
The receivable leg is always present (this endpoint is `/receivables/settle`).
The payable leg appears when the user has open expenses being netted against
the receivable. The credit leg appears when cash > settle target, absorbing
the overflow as a liability libra owes the user going forward.
Constraint enforced inline:
cash_paid_fiat = total_receivable_fiat - total_payable_fiat + credit_overflow_fiat
Args:
user_id: User ID
cash_account: Payment-method account name (e.g. "Assets:Cash")
receivable_account: User's receivable account being cleared
payable_account: User's payable account being cleared (omit when no payable)
credit_account: User's credit account receiving overflow (omit when no overflow)
cash_paid_fiat: What the user paid in cash, unsigned
total_receivable_fiat: Gross receivable being cleared (unsigned, 0 if none)
total_payable_fiat: Gross payable being cleared (unsigned, 0 if none)
credit_overflow_fiat: Excess cash going to credit (unsigned, 0 if none)
fiat_currency: Currency code (EUR, USD, etc.)
description: Entry narration
entry_date: Date of settlement
payment_method: cash / bank_transfer / check / other
reference: Optional caller-supplied reference (becomes an extra link)
settled_entry_links: Source entry links being cleared
(e.g. `["exp-abc", "rcv-def"]`). The audit trail for which
originals this settlement reconciles.
Returns:
Fava API entry dict ready for `fava.add_entry`.
Raises:
ValueError: if any amount is negative, or if the cash-balance
constraint above is not satisfied.
"""
for label, value in (
("cash_paid_fiat", cash_paid_fiat),
("total_receivable_fiat", total_receivable_fiat),
("total_payable_fiat", total_payable_fiat),
("credit_overflow_fiat", credit_overflow_fiat),
):
if value < 0:
raise ValueError(f"{label} must be non-negative; got {value}")
expected_cash = total_receivable_fiat - total_payable_fiat + credit_overflow_fiat
if abs(cash_paid_fiat - expected_cash) > Decimal("0.01"):
raise ValueError(
f"cash_paid_fiat {cash_paid_fiat} does not match expected "
f"{expected_cash} (= receivable {total_receivable_fiat} "
f"- payable {total_payable_fiat} + credit {credit_overflow_fiat})"
)
if total_payable_fiat > 0 and not payable_account:
raise ValueError("payable_account required when total_payable_fiat > 0")
if credit_overflow_fiat > 0 and not credit_account:
raise ValueError("credit_account required when credit_overflow_fiat > 0")
postings: List[Dict[str, Any]] = [
{"account": cash_account, "amount": f"{cash_paid_fiat:.2f} {fiat_currency}"},
{"account": receivable_account, "amount": f"-{total_receivable_fiat:.2f} {fiat_currency}"},
]
if total_payable_fiat > 0:
postings.append({
"account": payable_account,
"amount": f"{total_payable_fiat:.2f} {fiat_currency}",
})
if credit_overflow_fiat > 0:
postings.append({
"account": credit_account,
"amount": f"-{credit_overflow_fiat:.2f} {fiat_currency}",
})
payment_method_map = {
"cash": ("cash_settlement", "cash-payment"),
"bank_transfer": ("bank_settlement", "bank-transfer"),
"check": ("check_settlement", "check-payment"),
"btc_onchain": ("onchain_settlement", "onchain-payment"),
"other": ("manual_settlement", "manual-payment"),
}
source, tag = payment_method_map.get(
payment_method.lower(), ("manual_settlement", "manual-payment"),
)
entry_meta: Dict[str, Any] = {
"user-id": user_id,
"source": source,
"payment-type": "net-settlement",
}
links: List[str] = []
if settled_entry_links:
links.extend(settled_entry_links)
if reference:
links.append(sanitize_link(reference))
return format_transaction(
date_val=entry_date,
flag="*",
narration=description,
postings=postings,
tags=[tag, "settlement", "net-settlement"],
links=links,
meta=entry_meta,
)
def format_revenue_entry( def format_revenue_entry(
payment_account: str, payment_account: str,
revenue_account: str, revenue_account: str,

12
fava_client.py
View file

@ -820,15 +820,10 @@ class FavaClient:
# GROUP BY currency prevents mixing EUR and SATS face values in sum(number). # GROUP BY currency prevents mixing EUR and SATS face values in sum(number).
# sum(weight) gives SATS for both EUR @@ SATS entries and plain SATS entries. # sum(weight) gives SATS for both EUR @@ SATS entries and plain SATS entries.
# sum(number) on EUR rows gives the fiat amount; on SATS rows gives sats paid. # sum(number) on EUR rows gives the fiat amount; on SATS rows gives sats paid.
# Credit is the overpay-absorbing liability per libra-#41 — it lives
# on the same per-user namespace as Payable and contributes to the
# user's net obligation with the same sign as Payable (negative on
# Liabilities means libra owes user). Folding it into the same query
# means the displayed net always already accounts for credit.
query = f""" query = f"""
SELECT account, currency, sum(number), sum(weight) SELECT account, currency, sum(number), sum(weight)
WHERE account ~ ':User-{user_id_prefix}' WHERE account ~ ':User-{user_id_prefix}'
AND (account ~ 'Payable' OR account ~ 'Receivable' OR account ~ 'Credit') AND (account ~ 'Payable' OR account ~ 'Receivable')
AND flag = '*' AND flag = '*'
GROUP BY account, currency GROUP BY account, currency
""" """
@ -975,11 +970,10 @@ class FavaClient:
""" """
from decimal import Decimal from decimal import Decimal
# GROUP BY currency prevents mixing EUR and SATS face values in sum(number). # GROUP BY currency prevents mixing EUR and SATS face values in sum(number)
# Credit per libra-#41 — see get_user_balance_bql for the rationale.
query = """ query = """
SELECT account, currency, sum(number), sum(weight) SELECT account, currency, sum(number), sum(weight)
WHERE (account ~ 'Payable:User-' OR account ~ 'Receivable:User-' OR account ~ 'Credit:User-') WHERE (account ~ 'Payable:User-' OR account ~ 'Receivable:User-')
AND flag = '*' AND flag = '*'
GROUP BY account, currency GROUP BY account, currency
""" """

5
models.py
View file

@ -96,11 +96,6 @@ class UserBalance(BaseModel):
user_id: str user_id: str
balance: int # positive = libra owes user, negative = user owes libra balance: int # positive = libra owes user, negative = user owes libra
accounts: list[Account] = [] accounts: list[Account] = []
# Per-account breakdown surfaced from get_user_balance_bql so UIs (libra
# extension dashboard + webapp) can render Payable / Receivable / Credit
# as distinct line items. Each entry: {"account": str, "sats": int,
# "eur": Decimal}. Wired up for libra-#41's display contract.
account_balances: list[dict] = []
fiat_balances: dict[str, Decimal] = {} # e.g. {"EUR": Decimal("250.0"), "USD": Decimal("100.0")} fiat_balances: dict[str, Decimal] = {} # e.g. {"EUR": Decimal("250.0"), "USD": Decimal("100.0")}
# Lifetime totals (original entries only; not net of reconciliation) # Lifetime totals (original entries only; not net of reconciliation)
total_expenses_sats: int = 0 total_expenses_sats: int = 0

48
tests/README.md
View file

@ -1,48 +0,0 @@
# Libra extension tests
Integration tests covering the user- and admin-facing flows of the libra extension. Tests run against a real `fava` subprocess and a full LNbits app so they catch behaviour that mocks would miss (BQL semantics, Beancount arithmetic, multi-currency aggregation, HTTP boundary).
## Layout
- `conftest.py` — session-scoped Fava subprocess + LNbits app + user/wallet fixtures.
- `helpers.py` — high-level wrappers for the common API flows (`post_expense`, `settle_receivable`, `approve_manual_payment_request`, …). One per intention, so test bodies read as sequences of actions rather than HTTP calls.
- `test_smoke.py` — single end-to-end test; run first to validate the harness.
- `test_<area>_api.py` — per-flow coverage (entries, balances, settlement, manual payment requests, lightning, reconciliation, settings/auth, void/reject).
- `test_unit.py` — pure functions (`beancount_format`, `account_utils`, `core/validation`); no harness.
## Prerequisites
The harness requires `fava` on PATH. On NixOS:
```bash
nix-shell -p python3Packages.fava
```
Inside the regtest container `fava` is already provisioned.
## Running
From the LNbits source root (with the libra extension reachable via `LNBITS_EXTENSIONS_PATH` or symlinked into `lnbits/extensions/`):
```bash
# Whole suite
pytest path/to/libra/tests
# Smoke test only (validate the harness before running everything)
pytest path/to/libra/tests/test_smoke.py
# One area
pytest path/to/libra/tests/test_balances_api.py
# Single test, verbose
pytest path/to/libra/tests/test_balances_api.py::test_mixed_income_expense_nets_correctly -v
```
The Fava subprocess starts once per session (~1-2s) and is shared across tests; each test creates its own LNbits user so the shared ledger doesn't cause inter-test interference.
## Conventions
- **Tests assert intent, not shape.** Use the helpers in `helpers.py` for the request and assert on the *meaning* of the response (balance values, account names, settlement state), not on incidental keys in the JSON. This keeps tests resilient to non-behavioural API tweaks.
- **Currency-handling assertions use `pytest.approx`** for `Decimal`/`float` tolerance.
- **One canonical happy path per flow, plus boundary cases that matter** (voided entries excluded, pending entries excluded, cross-user isolation, auth gate rejection). Don't over-matrix.
- **Each test creates its own users** via the function-scoped `libra_user` / `libra_user_b` fixtures. The ledger is session-shared and accumulates entries; test isolation comes from unique user IDs, not ledger resets.

0
tests/__init__.py
View file

686
tests/conftest.py
View file

@ -1,686 +0,0 @@
"""Libra test infrastructure.
Brings up:
- A session-scoped Fava subprocess against a temp .beancount ledger
- A session-scoped LNbits FastAPI app with Libra extension activated
- The Libra FavaClient pointed at the test Fava instance
- Function-scoped user/wallet fixtures, plus a session-scoped superuser
Run from the LNbits source root::
PYTHONPATH=. pytest lnbits/extensions/libra/tests
Requires the `fava` binary on PATH. On NixOS::
nix-shell -p python3Packages.fava --run "pytest lnbits/extensions/libra/tests"
"""
import os
import tempfile
# IMPORTANT: configure the LNbits data folder BEFORE importing anything from
# lnbits. `lnbits/db.py` constructs Database instances at module-import time
# and freezes `settings.lnbits_data_folder` at that moment — overriding it in
# a fixture later is too late to redirect the SQLite files.
_SESSION_DATA_DIR = tempfile.mkdtemp(prefix="libra-lnbits-data-")
os.environ.setdefault("LNBITS_DATA_FOLDER", _SESSION_DATA_DIR)
# Lightning-invoice tests need a non-VoidWallet backend, but switching to
# FakeWallet here causes the LifespanManager teardown to hang indefinitely
# (the Lightning subsystem's background tasks don't unwind cleanly under
# anyio's TestRunner). Keeping VoidWallet — Lightning-invoice-generation
# tests are marked `skip` until a separate LN-harness strategy lands.
import asyncio # noqa: E402
import copy # noqa: E402
import inspect # noqa: E402
import shutil # noqa: E402
import socket # noqa: E402
import subprocess # noqa: E402
import time # noqa: E402
from pathlib import Path # noqa: E402
from typing import AsyncIterator, Iterator # noqa: E402
from uuid import uuid4 # noqa: E402
import httpx
import pytest
from asgi_lifespan import LifespanManager
from httpx import ASGITransport, AsyncClient
from lnbits.app import create_app
from lnbits.core.crud import (
create_wallet,
delete_account,
get_user,
)
from lnbits.core.models.users import UpdateSuperuserPassword
from lnbits.core.services import create_user_account
from lnbits.core.views.auth_api import first_install
from lnbits.settings import AuthMethods, EditableSettings, Settings
from lnbits.settings import settings as lnbits_settings
LEDGER_SLUG = "libra-test"
# ---------------------------------------------------------------------------
# Settings overrides
# ---------------------------------------------------------------------------
_PURE_SETTINGS = copy.deepcopy(lnbits_settings)
_PURE_SETTINGS_FIELDS = tuple(
sorted(
{
f
for f in Settings.readonly_fields()
if f != "super_user"
}
| {
name
for name in inspect.signature(EditableSettings).parameters
if not name.startswith("_")
}
)
)
def _settings_cleanup(settings: Settings) -> None:
"""Reset mutable settings to their pre-test snapshot, then re-apply
test-specific overrides on top so each test starts from the same baseline.
Mirrors the shape of lnbits/main/tests/conftest.py: restore PURE, then
set the values the tests rely on. Without this, autouse cleanup wipes
out everything the session-scoped `settings` fixture set up.
"""
for field in _PURE_SETTINGS_FIELDS:
setattr(settings, field, getattr(_PURE_SETTINGS, field))
# Test-specific overrides — these must survive cleanup between tests.
settings.auth_https_only = False
settings.lnbits_data_folder = _SESSION_DATA_DIR
settings.lnbits_admin_extensions = [] # libra is a multi-user extension, not admin-only
settings.lnbits_admin_ui = True
settings.lnbits_extensions_default_install = []
settings.lnbits_extensions_deactivate_all = False
settings.lnbits_allow_new_accounts = True
settings.lnbits_allowed_users = []
settings.auth_allowed_methods = AuthMethods.all()
settings.auth_credetials_update_threshold = 120
settings.lnbits_require_user_activation = False
settings.lnbits_user_activation_by_invitation_code = False
settings.lnbits_register_reusable_activation_code = ""
settings.lnbits_register_one_time_activation_codes = []
@pytest.fixture(scope="session")
def anyio_backend() -> str:
return "asyncio"
@pytest.fixture(scope="session")
def settings() -> Iterator[Settings]:
"""LNbits settings configured for the libra test session.
Mirrors lnbits/main/tests/conftest.py: do NOT pre-set super_user; the boot
sequence assigns a UUID and creates the matching account. The `super_user`
fixture reads settings.super_user after first_install completes.
The data folder was set via LNBITS_DATA_FOLDER at the top of this module
so the lnbits/db.py import-time directory creation lands in the right
place; nothing to do here except make sure it stays consistent.
"""
lnbits_settings.auth_https_only = False
lnbits_settings.lnbits_admin_extensions = ["libra"]
lnbits_settings.lnbits_data_folder = _SESSION_DATA_DIR
lnbits_settings.lnbits_admin_ui = True
lnbits_settings.lnbits_extensions_default_install = []
lnbits_settings.lnbits_extensions_deactivate_all = False
yield lnbits_settings
@pytest.fixture(autouse=True)
def _per_test_settings_reset(settings: Settings) -> Iterator[None]:
_settings_cleanup(settings)
yield
_settings_cleanup(settings)
# ---------------------------------------------------------------------------
# Fava subprocess
# ---------------------------------------------------------------------------
def _find_free_port() -> int:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
sock.bind(("127.0.0.1", 0))
return sock.getsockname()[1]
MINIMAL_LEDGER = """; Test ledger for Libra extension integration tests
; Title must slugify to match LEDGER_SLUG — Fava derives the URL slug from this.
option "title" "libra-test"
option "operating_currency" "EUR"
option "operating_currency" "SATS"
option "render_commas" "TRUE"
2020-01-01 commodity EUR
2020-01-01 commodity SATS
2020-01-01 open Assets:Lightning:Balance EUR,SATS
2020-01-01 open Assets:Bitcoin:Lightning EUR,SATS
2020-01-01 open Assets:Cash EUR,SATS
2020-01-01 open Equity:Opening-Balances EUR,SATS
2020-01-01 open Income:Generic EUR,SATS
2020-01-01 open Expenses:Generic EUR,SATS
"""
@pytest.fixture(scope="session")
def fava_ledger_path(tmp_path_factory: pytest.TempPathFactory) -> Path:
"""Session-scoped .beancount file Fava reads from."""
ledger_dir = tmp_path_factory.mktemp("libra-ledger")
ledger = ledger_dir / f"{LEDGER_SLUG}.beancount"
ledger.write_text(MINIMAL_LEDGER)
return ledger
@pytest.fixture(scope="session")
def fava_process(fava_ledger_path: Path) -> Iterator[str]:
"""Spawn fava as a subprocess, yield its base URL, terminate on teardown."""
fava_bin = shutil.which("fava")
if not fava_bin:
pytest.skip(
"fava not found on PATH; "
"install with `pip install fava` or `nix-shell -p python3Packages.fava`"
)
port = _find_free_port()
base_url = f"http://127.0.0.1:{port}"
proc = subprocess.Popen(
[
fava_bin,
"--host", "127.0.0.1",
"--port", str(port),
str(fava_ledger_path),
],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
env={**os.environ, "BEANCOUNT_FILE": str(fava_ledger_path)},
)
deadline = time.monotonic() + 15.0
ready = False
while time.monotonic() < deadline:
if proc.poll() is not None:
raise RuntimeError(
f"fava exited early with returncode {proc.returncode}"
)
try:
r = httpx.get(f"{base_url}/{LEDGER_SLUG}/api/changed", timeout=0.5)
if r.status_code == 200:
ready = True
break
except httpx.RequestError:
pass
time.sleep(0.1)
if not ready:
proc.terminate()
raise RuntimeError("fava did not become ready within 15s")
try:
yield base_url
finally:
proc.terminate()
try:
proc.wait(timeout=5)
except subprocess.TimeoutExpired:
proc.kill()
# ---------------------------------------------------------------------------
# LNbits app + Libra extension
# ---------------------------------------------------------------------------
def _import_libra(submodule: str):
"""Import a libra submodule under whichever path the active LNbits setup uses.
LNbits resolves an extension's module name dynamically: `lnbits.extensions.<ext>`
when extensions live in the default `lnbits/extensions/` directory, or just
`<ext>` when `LNBITS_EXTENSIONS_PATH` points elsewhere. Tests should work in
both setups.
"""
import importlib
for prefix in ("lnbits.extensions.libra", "libra"):
try:
return importlib.import_module(f"{prefix}.{submodule}")
except ModuleNotFoundError:
continue
raise ModuleNotFoundError(
f"libra.{submodule}: tried 'lnbits.extensions.libra.{submodule}' and "
f"'libra.{submodule}'. Is LNBITS_EXTENSIONS_PATH pointing at the libra parent dir, "
f"or is libra symlinked into lnbits/extensions/?"
)
async def _enable_libra_for_user(user_id: str) -> None:
"""Set libra to active in the user_extensions table for `user_id`.
LNbits gates every extension API path through `check_user_extension_access`,
which requires the calling user to have the extension marked active in
`user_extensions`. New accounts have no extensions enabled, so the API
rejects them with 403 until we flip the row.
"""
from lnbits.core.services.users import update_user_extensions
await update_user_extensions(user_id, ["libra"])
async def _activate_libra(fava_url: str, super_user_id: str) -> None:
"""Point libra at the test Fava instance and enable it for the superuser.
Libra is auto-discovered + auto-installed at LNbits boot via
`LNBITS_EXTENSIONS_PATH`, so its router is already mounted, migrations
already ran, and `libra_start()` already initialised a FavaClient with
the default `http://localhost:3333/libra-ledger` URL. Three things still
need doing:
1. Redirect the FavaClient at the test Fava instance.
2. Persist the override in `extension_settings` so any caller that goes
through `services.get_settings()` picks it up too.
3. Enable libra for the superuser — per-user activation isn't automatic.
"""
libra_fava_client = _import_libra("fava_client")
libra_crud = _import_libra("crud")
libra_fava_client.init_fava_client(
fava_url=fava_url,
ledger_slug=LEDGER_SLUG,
timeout=5.0,
)
await libra_crud.db.execute("DELETE FROM extension_settings")
await libra_crud.db.execute(
"""
INSERT INTO extension_settings (id, fava_url, fava_ledger_slug, fava_timeout)
VALUES (:id, :fava_url, :slug, :timeout)
""",
{
"id": uuid4().hex,
"fava_url": fava_url,
"slug": LEDGER_SLUG,
"timeout": 5.0,
},
)
await _enable_libra_for_user(super_user_id)
@pytest.fixture(scope="session")
async def app(settings: Settings, fava_process: str) -> AsyncIterator:
"""Session-scoped LNbits app with Libra activated."""
app = create_app()
# First-time startup runs all core + libra migrations (~3-5s on cold disk),
# plus libra_start() initialises the Fava client and background tasks.
# Bump the timeout well above asgi_lifespan's 10s default so a slow
# migration step or Fava startup race doesn't spuriously fail the session.
async with LifespanManager(app, startup_timeout=60, shutdown_timeout=20) as manager:
settings.first_install = True
# pragma: allowlist secret start
await first_install(
UpdateSuperuserPassword(
username="superadmin",
password="secret1234",
password_repeat="secret1234",
first_install_token=settings.first_install_token,
)
# pragma: allowlist secret end
)
await _activate_libra(
fava_url=fava_process,
super_user_id=settings.super_user,
)
yield manager.app
@pytest.fixture(scope="session")
async def client(app, settings: Settings) -> AsyncIterator[AsyncClient]:
url = f"http://{settings.host}:{settings.port}"
async with AsyncClient(transport=ASGITransport(app=app), base_url=url) as client:
yield client
# ---------------------------------------------------------------------------
# Users
# ---------------------------------------------------------------------------
@pytest.fixture(scope="session")
async def super_user(app, settings: Settings):
"""The superadmin account created by first_install."""
# first_install sets settings.super_user to the actual ID it created.
user = await get_user(settings.super_user)
assert user is not None, "superadmin was not created by first_install"
return user
@pytest.fixture
async def libra_user(app):
"""A fresh non-admin user with a wallet. Function-scoped — each test gets its own.
Libra is enabled in the user_extensions table for this user so the API
doesn't 403 with "Extension 'libra' not enabled."
"""
user = await create_user_account()
wallet = await create_wallet(
user_id=user.id, wallet_name=f"libra-test-{uuid4().hex[:6]}"
)
await _enable_libra_for_user(user.id)
yield user, wallet
# Cleanup: best-effort
try:
await delete_account(user.id)
except Exception:
pass
@pytest.fixture
async def libra_user_b(app):
"""A second fresh non-admin user, for tests that need cross-user assertions."""
user = await create_user_account()
wallet = await create_wallet(
user_id=user.id, wallet_name=f"libra-test-{uuid4().hex[:6]}"
)
await _enable_libra_for_user(user.id)
yield user, wallet
try:
await delete_account(user.id)
except Exception:
pass
# ---------------------------------------------------------------------------
# Auth headers
# ---------------------------------------------------------------------------
async def _user_bearer(client: AsyncClient, user_id: str) -> dict:
"""Bearer headers for a non-admin user via the `/auth/usr` user-id-only flow.
Admin/super accounts are blocked from this flow (LNbits forces them to
use username+password); regular users use it freely. Required for libra
endpoints that depend on `check_user_exists` (Bearer/cookie/usr) rather
than on a wallet API key.
"""
r = await client.post("/api/v1/auth/usr", json={"usr": user_id})
client.cookies.clear()
token = r.json().get("access_token")
assert token, f"user-id login failed: {r.status_code} {r.text}"
return {
"Authorization": f"Bearer {token}",
"Content-type": "application/json",
}
async def _superadmin_bearer(client: AsyncClient) -> dict:
"""Bearer headers for the superadmin via username+password auth.
`/api/v1/auth/usr` (user-id-only auth) is rejected for admin users —
LNbits enforces username+password for accounts in `lnbits_admin_users`
or the super_user account. So super-user fixtures use the username
flow that `first_install` configured.
"""
r = await client.post(
"/api/v1/auth", json={"username": "superadmin", "password": "secret1234"}
)
client.cookies.clear()
token = r.json().get("access_token")
assert token, f"superadmin login failed: {r.status_code} {r.text}"
return {
"Authorization": f"Bearer {token}",
"Content-type": "application/json",
}
@pytest.fixture
async def super_user_bearer_headers(client: AsyncClient, super_user) -> dict:
"""Bearer headers for the few endpoints that use LNbits `check_super_user`.
The `/libra/api/v1/settings` endpoints (and other libra paths that take
`User = Depends(check_super_user)`) require a Bearer token from
username+password login. Most other libra admin endpoints use the
wallet-admin-key auth flow — use `super_user_headers` for those.
"""
return await _superadmin_bearer(client)
@pytest.fixture
async def super_user_headers(super_user, libra_wallet) -> dict:
"""Admin-key headers for libra admin endpoints that use the wallet auth flow.
Libra's `require_super_user` dependency takes a `WalletTypeInfo` via
`require_admin_key` and verifies the wallet's owner is the LNbits
super user. So we authenticate by sending the super-user-owned wallet's
admin key as `X-Api-Key`.
"""
return admin_key_headers(libra_wallet)
def invoice_key_headers(wallet) -> dict:
"""Wallet invoice-key headers (X-Api-Key) — for require_invoice_key endpoints."""
return {"X-Api-Key": wallet.inkey, "Content-type": "application/json"}
def admin_key_headers(wallet) -> dict:
"""Wallet admin-key headers (X-Api-Key) — for require_admin_key endpoints."""
return {"X-Api-Key": wallet.adminkey, "Content-type": "application/json"}
# ---------------------------------------------------------------------------
# Libra-specific session setup: wallet, accounts
# ---------------------------------------------------------------------------
@pytest.fixture(scope="session")
async def libra_wallet(
app, settings: Settings, super_user, fava_process: str, client: AsyncClient,
):
"""Session-scoped: create a wallet for the super user and register it
as the libra wallet in extension_settings.
Most flows (expense, income, settle, pay-user) refuse to operate until
this is set. Session-scoped because it's a one-time setup that any test
can share.
"""
wallet = await create_wallet(user_id=super_user.id, wallet_name="libra-main")
# Configure libra_wallet_id via the settings API so the in-memory cache
# (services.update_settings) refreshes too.
#
# Critical: include fava_url + fava_ledger_slug in the body so that
# services.update_settings()'s re-init of the FavaClient doesn't reset
# us to the default `http://localhost:3333/libra-ledger`. The settings
# endpoint rewrites the global FavaClient from the body's contents on
# every call.
headers = await _superadmin_bearer(client)
r = await client.put(
"/libra/api/v1/settings",
headers=headers,
json={
"libra_wallet_id": wallet.id,
"fava_url": fava_process,
"fava_ledger_slug": LEDGER_SLUG,
},
)
assert r.status_code == 200, f"libra_wallet setup failed: {r.status_code} {r.text}"
return wallet
@pytest.fixture(scope="session")
async def standard_accounts(app, super_user, libra_wallet, client: AsyncClient):
"""Session-scoped: create a small set of accounts used across tests.
Returns a dict of {short_name: account_dict}. Each account has at least
`id` and `name` keys.
"""
# `/accounts` POST is gated by `require_super_user` (libra-level, wallet
# admin-key flow), so we authenticate with the super-user's wallet key.
headers = admin_key_headers(libra_wallet)
async def _list_lookup(name: str) -> dict | None:
r = await client.get("/libra/api/v1/accounts", headers=headers)
if r.status_code != 200:
return None
for a in r.json():
if a.get("name") == name:
return a
return None
async def _create(name: str, account_type: str) -> dict:
# Get-then-create. Some accounts (Assets:Cash, etc.) are auto-synced
# into libra's DB from the Beancount Open directives by the account-sync
# background task. Posting a duplicate raises IntegrityError → 500;
# checking first avoids the race and the noisy error log.
existing = await _list_lookup(name)
if existing:
return existing
r = await client.post(
"/libra/api/v1/accounts",
headers=headers,
json={"name": name, "account_type": account_type},
)
if r.status_code == 201:
return r.json()
# Lost the race between our GET and POST — sync ran in between.
existing = await _list_lookup(name)
if existing:
return existing
raise AssertionError(f"create account {name}: {r.status_code} {r.text}")
return {
"expense_food": await _create("Expenses:Test:Food", "expense"),
"expense_supplies": await _create("Expenses:Test:Supplies", "expense"),
"revenue_rent": await _create("Income:Test:Rent", "revenue"),
"revenue_fees": await _create("Income:Test:Fees", "revenue"),
# Cash for revenue/settlement payment-method tests. Already declared
# as an Open directive in the Beancount file (see MINIMAL_LEDGER),
# but needs a libra-DB row too because the revenue endpoint validates
# payment-method-account via libra's local lookup.
"assets_cash": await _create("Assets:Cash", "asset"),
# Lightning balance account — the manual-payment-request approve
# endpoint posts the payment leg against this. Open directive lives
# in MINIMAL_LEDGER's Assets:Lightning:Balance, but the API code
# looks up `Assets:Bitcoin:Lightning` specifically.
"assets_lightning": await _create("Assets:Bitcoin:Lightning", "asset"),
}
# ---------------------------------------------------------------------------
# Configured user — wallet set + can submit expenses to the standard accounts
# ---------------------------------------------------------------------------
async def _grant_account_permissions(
client: AsyncClient,
libra_wallet,
user_id: str,
grants: list[tuple[str, str]],
) -> None:
"""Grant a list of (account_id, permission_type) pairs to a user.
Existing perms come back as 409; that's idempotent for fixture re-runs.
"""
headers = admin_key_headers(libra_wallet)
for account_id, permission_type in grants:
r = await client.post(
"/libra/api/v1/admin/permissions",
headers=headers,
json={
"user_id": user_id,
"account_id": account_id,
"permission_type": permission_type,
},
)
# 201 created; 409 if it already existed (idempotent).
assert r.status_code in (200, 201, 409), (
f"grant permission failed: {r.status_code} {r.text}"
)
@pytest.fixture
async def configured_user(
app, super_user, libra_wallet, standard_accounts, client: AsyncClient,
):
"""Function-scoped: fresh user with a wallet, configured for libra,
permitted to submit expenses to the standard test accounts.
Yields (user, wallet) ready to make any user-facing API call.
"""
user = await create_user_account()
wallet = await create_wallet(
user_id=user.id, wallet_name=f"libra-test-{uuid4().hex[:6]}"
)
await _enable_libra_for_user(user.id)
# User registers their own wallet with libra. The endpoint uses
# `check_user_exists` which accepts either a Bearer access token OR
# a `?usr=<id>` query param — we use the query param to avoid the
# cookie-state interleaving that bites when two configured_user
# fixtures stack in the same test.
r = await client.put(
f"/libra/api/v1/user/wallet?usr={user.id}",
json={"user_wallet_id": wallet.id},
)
assert r.status_code == 200, f"user wallet setup failed: {r.status_code} {r.text}"
# Grant submit_expense on every expense account, submit_income on every
# revenue account, so tests can hit either user-side entry endpoint.
grants = [
(a["id"], "submit_expense")
for k, a in standard_accounts.items() if k.startswith("expense_")
] + [
(a["id"], "submit_income")
for k, a in standard_accounts.items() if k.startswith("revenue_")
]
await _grant_account_permissions(client, libra_wallet, user.id, grants)
yield user, wallet
try:
await delete_account(user.id)
except Exception:
pass
@pytest.fixture
async def configured_user_b(
app, super_user, libra_wallet, standard_accounts, client: AsyncClient,
):
"""A second configured user for cross-user tests."""
user = await create_user_account()
wallet = await create_wallet(
user_id=user.id, wallet_name=f"libra-test-{uuid4().hex[:6]}"
)
await _enable_libra_for_user(user.id)
r = await client.put(
f"/libra/api/v1/user/wallet?usr={user.id}",
json={"user_wallet_id": wallet.id},
)
assert r.status_code == 200, f"user wallet setup failed: {r.status_code} {r.text}"
grants = [
(a["id"], "submit_expense")
for k, a in standard_accounts.items() if k.startswith("expense_")
] + [
(a["id"], "submit_income")
for k, a in standard_accounts.items() if k.startswith("revenue_")
]
await _grant_account_permissions(client, libra_wallet, user.id, grants)
yield user, wallet
try:
await delete_account(user.id)
except Exception:
pass

392
tests/helpers.py
View file

@ -1,392 +0,0 @@
"""Convenience helpers for Libra integration tests.
Wrap the most common multi-step flows so each test reads as a sequence of
intentions rather than as a sequence of HTTP calls. Every helper returns the
parsed JSON response and asserts a successful status code — tests that want
to assert on failures should call the endpoint directly.
All amounts are passed as Decimal (or numeric string). Currency goes as a
separate ISO code field — this matches `models.ExpenseEntry` / `ReceivableEntry`
/ `SettleReceivable` / `PayUser` etc., which all carry `amount: Decimal` and
`currency: Optional[str]` independently.
"""
from decimal import Decimal
from typing import Any, Optional, Union
from httpx import AsyncClient
Amount = Union[Decimal, int, float, str]
def _amount(value: Amount) -> str:
"""Coerce amount to a JSON-serialisable string Pydantic will parse as Decimal."""
return str(value)
# ---------------------------------------------------------------------------
# Setup — libra wallet + per-user wallet + accounts + permissions
# ---------------------------------------------------------------------------
async def configure_libra_wallet(
client: AsyncClient,
*,
super_user_headers: dict,
libra_wallet_id: str,
) -> dict:
"""Super user sets the libra wallet (required before any entry endpoint works)."""
r = await client.put(
"/libra/api/v1/settings",
headers=super_user_headers,
json={"libra_wallet_id": libra_wallet_id},
)
assert r.status_code == 200, f"configure_libra_wallet failed: {r.status_code} {r.text}"
return r.json()
async def configure_user_wallet(
client: AsyncClient,
*,
wallet_inkey: str,
user_wallet_id: str,
) -> dict:
"""User sets their personal wallet (required before they can submit entries)."""
r = await client.put(
"/libra/api/v1/user/wallet",
headers={"X-Api-Key": wallet_inkey, "Content-type": "application/json"},
json={"user_wallet_id": user_wallet_id},
)
assert r.status_code == 200, f"configure_user_wallet failed: {r.status_code} {r.text}"
return r.json()
async def create_account(
client: AsyncClient,
*,
super_user_headers: dict,
name: str,
account_type: str,
description: Optional[str] = None,
) -> dict:
"""Super user creates an account in the libra local DB.
`account_type` is one of "asset", "liability", "equity", "revenue", "expense".
"""
r = await client.post(
"/libra/api/v1/accounts",
headers=super_user_headers,
json={
"name": name,
"account_type": account_type,
"description": description,
},
)
assert r.status_code == 201, f"create_account failed: {r.status_code} {r.text}"
return r.json()
async def grant_permission(
client: AsyncClient,
*,
super_user_headers: dict,
user_id: str,
account_id: str,
permission_type: str = "submit_expense",
) -> dict:
r = await client.post(
"/libra/api/v1/admin/permissions",
headers=super_user_headers,
json={
"user_id": user_id,
"account_id": account_id,
"permission_type": permission_type,
},
)
assert r.status_code == 201, f"grant_permission failed: {r.status_code} {r.text}"
return r.json()
# ---------------------------------------------------------------------------
# Entries — user side
# ---------------------------------------------------------------------------
async def post_expense(
client: AsyncClient,
*,
wallet_inkey: str,
user_wallet_id: str,
amount: Amount,
description: str,
expense_account: str,
currency: Optional[str] = "EUR",
is_equity: bool = False,
) -> dict[str, Any]:
"""User submits an expense — creates Liability (libra owes user) or Equity contribution.
Returns the created JournalEntry payload.
"""
r = await client.post(
"/libra/api/v1/entries/expense",
headers={"X-Api-Key": wallet_inkey, "Content-type": "application/json"},
json={
"description": description,
"amount": _amount(amount),
"expense_account": expense_account,
"user_wallet": user_wallet_id,
"currency": currency,
"is_equity": is_equity,
},
)
assert r.status_code == 201, f"post_expense failed: {r.status_code} {r.text}"
return r.json()
async def post_income(
client: AsyncClient,
*,
wallet_inkey: str,
amount: Amount,
description: str,
revenue_account: str,
currency: str = "EUR",
) -> dict[str, Any]:
"""User submits income on libra's behalf — creates Receivable (user owes libra)."""
r = await client.post(
"/libra/api/v1/entries/income",
headers={"X-Api-Key": wallet_inkey, "Content-type": "application/json"},
json={
"description": description,
"amount": _amount(amount),
"revenue_account": revenue_account,
"currency": currency,
},
)
assert r.status_code == 201, f"post_income failed: {r.status_code} {r.text}"
return r.json()
async def list_user_entries(client: AsyncClient, *, wallet_inkey: str) -> list[dict]:
r = await client.get(
"/libra/api/v1/entries/user",
headers={"X-Api-Key": wallet_inkey},
)
assert r.status_code == 200, f"list_user_entries failed: {r.status_code} {r.text}"
return r.json()
# ---------------------------------------------------------------------------
# Entries — admin side
# ---------------------------------------------------------------------------
async def post_receivable(
client: AsyncClient,
*,
super_user_headers: dict,
user_id: str,
amount: Amount,
description: str,
revenue_account: str,
currency: str = "EUR",
) -> dict[str, Any]:
"""Admin records a receivable — user owes libra."""
r = await client.post(
"/libra/api/v1/entries/receivable",
headers=super_user_headers,
json={
"user_id": user_id,
"amount": _amount(amount),
"description": description,
"revenue_account": revenue_account,
"currency": currency,
},
)
assert r.status_code == 201, f"post_receivable failed: {r.status_code} {r.text}"
return r.json()
async def post_revenue(
client: AsyncClient,
*,
super_user_headers: dict,
amount: Amount,
description: str,
revenue_account: str,
payment_method_account: str,
currency: str = "EUR",
) -> dict[str, Any]:
r = await client.post(
"/libra/api/v1/entries/revenue",
headers=super_user_headers,
json={
"amount": _amount(amount),
"description": description,
"revenue_account": revenue_account,
"payment_method_account": payment_method_account,
"currency": currency,
},
)
assert r.status_code == 201, f"post_revenue failed: {r.status_code} {r.text}"
return r.json()
# ---------------------------------------------------------------------------
# Balances
# ---------------------------------------------------------------------------
async def get_balance(client: AsyncClient, *, wallet_inkey: str) -> dict[str, Any]:
"""Calling user's balance (or libra total if invoked by super user)."""
r = await client.get(
"/libra/api/v1/balance",
headers={"X-Api-Key": wallet_inkey},
)
assert r.status_code == 200, f"get_balance failed: {r.status_code} {r.text}"
return r.json()
async def get_all_balances(
client: AsyncClient, *, super_user_headers: dict
) -> list[dict]:
r = await client.get(
"/libra/api/v1/balances/all",
headers=super_user_headers,
)
assert r.status_code == 200, f"get_all_balances failed: {r.status_code} {r.text}"
return r.json()
# ---------------------------------------------------------------------------
# Settlement
# ---------------------------------------------------------------------------
async def settle_receivable(
client: AsyncClient,
*,
super_user_headers: dict,
user_id: str,
amount: Amount,
description: str = "Cash settlement",
payment_method: str = "cash",
currency: str = "EUR",
) -> dict[str, Any]:
"""Admin records that user paid libra (e.g. cash, bank transfer)."""
r = await client.post(
"/libra/api/v1/receivables/settle",
headers=super_user_headers,
json={
"user_id": user_id,
"amount": _amount(amount),
"description": description,
"payment_method": payment_method,
"currency": currency,
},
)
assert r.status_code == 200, f"settle_receivable failed: {r.status_code} {r.text}"
return r.json()
async def pay_user(
client: AsyncClient,
*,
super_user_headers: dict,
user_id: str,
amount: Amount,
description: str = "Libra pays user",
payment_method: str = "cash",
currency: str = "EUR",
) -> dict[str, Any]:
"""Admin records that libra paid user (e.g. cash, bank, lightning)."""
r = await client.post(
"/libra/api/v1/payables/pay",
headers=super_user_headers,
json={
"user_id": user_id,
"amount": _amount(amount),
"description": description,
"payment_method": payment_method,
"currency": currency,
},
)
assert r.status_code == 200, f"pay_user failed: {r.status_code} {r.text}"
return r.json()
# ---------------------------------------------------------------------------
# Manual payment requests
# ---------------------------------------------------------------------------
async def submit_manual_payment_request(
client: AsyncClient,
*,
wallet_inkey: str,
amount_sats: int,
description: str,
) -> dict[str, Any]:
"""User asks for libra to pay them via a manual (non-Lightning) route.
Body matches `CreateManualPaymentRequest`: amount in satoshis (no fiat
conversion at this endpoint), description for the admin to review.
"""
r = await client.post(
"/libra/api/v1/manual-payment-request",
headers={"X-Api-Key": wallet_inkey, "Content-type": "application/json"},
json={"amount": amount_sats, "description": description},
)
assert r.status_code in (200, 201), (
f"submit_manual_payment_request failed: {r.status_code} {r.text}"
)
return r.json()
async def approve_manual_payment_request(
client: AsyncClient, *, super_user_headers: dict, request_id: str,
) -> dict[str, Any]:
r = await client.post(
f"/libra/api/v1/manual-payment-requests/{request_id}/approve",
headers=super_user_headers,
)
assert r.status_code == 200, (
f"approve_manual_payment_request failed: {r.status_code} {r.text}"
)
return r.json()
async def approve_entry(
client: AsyncClient, *, super_user_headers: dict, entry_id: str,
) -> dict[str, Any]:
"""Admin approves a pending journal entry, flipping its flag from `!` to `*`."""
r = await client.post(
f"/libra/api/v1/entries/{entry_id}/approve",
headers=super_user_headers,
)
assert r.status_code == 200, f"approve_entry failed: {r.status_code} {r.text}"
return r.json()
async def reject_entry(
client: AsyncClient, *, super_user_headers: dict, entry_id: str,
) -> dict[str, Any]:
"""Admin rejects a pending journal entry, marking it #voided."""
r = await client.post(
f"/libra/api/v1/entries/{entry_id}/reject",
headers=super_user_headers,
)
assert r.status_code == 200, f"reject_entry failed: {r.status_code} {r.text}"
return r.json()
async def reject_manual_payment_request(
client: AsyncClient, *, super_user_headers: dict, request_id: str,
) -> dict[str, Any]:
r = await client.post(
f"/libra/api/v1/manual-payment-requests/{request_id}/reject",
headers=super_user_headers,
)
assert r.status_code == 200, (
f"reject_manual_payment_request failed: {r.status_code} {r.text}"
)
return r.json()

452
tests/test_balances_api.py
View file

@ -1,452 +0,0 @@
"""Balance display tests — the user-named "mixture of income and expenses
displayed correctly" scenario.
The balance API returns figures from libra's perspective:
- Negative `fiat_balances[CCY]` → libra owes the user
- Positive `fiat_balances[CCY]` → user owes libra
- Sum across Payable + Receivable + Credit per currency
(Credit added per libra-#41: overpayment lands as a liability that
libra owes the user going forward, naturally subtracting from net.)
Lifetime totals (`total_expenses_fiat`, `total_income_fiat`) are kept
separate per the `models.py:93` comment — "original entries only; not net of
reconciliation" — so they don't reflect settlement activity or credit.
Excluded from the balance query: pending entries (flag `!`), voided entries
(tag `voided`). Tested explicitly here so the contract is locked in.
Note: this file does NOT cover post-settlement netting; that's blocked on
issue #33 (settlement leaves both per-user accounts non-zero) and lives in
the settlement test file.
"""
import importlib
from datetime import date
from uuid import uuid4
import pytest
from .helpers import (
approve_entry,
get_all_balances,
get_balance,
list_user_entries,
post_expense,
post_income,
post_receivable,
reject_entry,
)
def _libra_module(submodule: str):
"""Import a libra submodule via whichever path the harness uses (matches
the resolver in conftest.py)."""
for prefix in ("lnbits.extensions.libra", "libra"):
try:
return importlib.import_module(f"{prefix}.{submodule}")
except ModuleNotFoundError:
continue
raise ModuleNotFoundError(f"libra.{submodule}")
async def _approve_and_refresh(client, wallet, super_user_headers, entry_id):
"""Approve a pending entry then force a fresh Fava read.
Workaround for libra issue #37 — BQL balance reads can lag add_entry
by a few ms. The user-journal endpoint forces a Fava reload.
"""
await approve_entry(
client, super_user_headers=super_user_headers, entry_id=entry_id,
)
await list_user_entries(client, wallet_inkey=wallet.inkey)
# ---------------------------------------------------------------------------
# Single-direction balances
# ---------------------------------------------------------------------------
@pytest.mark.anyio
async def test_pure_expense_balance_is_negative(
client, super_user_headers, configured_user, standard_accounts,
):
"""User submits a single expense → libra owes them → balance < 0 EUR."""
_, wallet = configured_user
entry = await post_expense(
client,
wallet_inkey=wallet.inkey,
user_wallet_id=wallet.id,
amount="40.00", currency="EUR",
description=f"Pure expense {uuid4().hex[:6]}",
expense_account=standard_accounts["expense_food"]["name"],
)
await _approve_and_refresh(client, wallet, super_user_headers, entry["id"])
balance = await get_balance(client, wallet_inkey=wallet.inkey)
eur = balance.get("fiat_balances", {}).get("EUR")
assert float(eur) == pytest.approx(-40.0), (
f"expected -40 EUR (libra owes user), got {eur}"
)
@pytest.mark.anyio
async def test_pure_income_balance_is_positive(
client, super_user_headers, configured_user, standard_accounts,
):
"""User submits a single income → user owes libra → balance > 0 EUR.
`/entries/income` records that the user collected money on libra's
behalf, creating an `Assets:Receivable:User-{id}` debit until they
settle by handing the cash over.
"""
_, wallet = configured_user
entry = await post_income(
client,
wallet_inkey=wallet.inkey,
amount="120.00", currency="EUR",
description=f"Pure income {uuid4().hex[:6]}",
revenue_account=standard_accounts["revenue_rent"]["name"],
)
await _approve_and_refresh(client, wallet, super_user_headers, entry["id"])
balance = await get_balance(client, wallet_inkey=wallet.inkey)
eur = balance.get("fiat_balances", {}).get("EUR")
assert float(eur) == pytest.approx(120.0), (
f"expected +120 EUR (user owes libra), got {eur}"
)
# ---------------------------------------------------------------------------
# Mixed direction — the headline scenario
# ---------------------------------------------------------------------------
@pytest.mark.anyio
async def test_mixed_expense_and_income_nets_correctly(
client, super_user_headers, configured_user, standard_accounts,
):
"""User has 50 EUR expense + 120 EUR income (both approved) → net
balance is +70 EUR (user owes libra 70).
This is the user's headline "displayed correctly" scenario — the
Payable and Receivable rows sum into one EUR figure.
"""
_, wallet = configured_user
expense = await post_expense(
client,
wallet_inkey=wallet.inkey,
user_wallet_id=wallet.id,
amount="50.00", currency="EUR",
description=f"Coffee {uuid4().hex[:6]}",
expense_account=standard_accounts["expense_food"]["name"],
)
income = await post_income(
client,
wallet_inkey=wallet.inkey,
amount="120.00", currency="EUR",
description=f"Cash deposit {uuid4().hex[:6]}",
revenue_account=standard_accounts["revenue_rent"]["name"],
)
await _approve_and_refresh(client, wallet, super_user_headers, expense["id"])
await _approve_and_refresh(client, wallet, super_user_headers, income["id"])
balance = await get_balance(client, wallet_inkey=wallet.inkey)
eur = balance.get("fiat_balances", {}).get("EUR")
assert float(eur) == pytest.approx(70.0), (
f"expected +70 EUR (120 - 50, user-owes-libra), got {eur} from {balance}"
)
@pytest.mark.anyio
async def test_mixed_expense_and_receivable_nets_correctly(
client, super_user_headers, configured_user, standard_accounts,
):
"""Admin-recorded receivable + user-submitted expense should net the
same way as expense + income — both push the receivable side."""
user, wallet = configured_user
await post_receivable(
client,
super_user_headers=super_user_headers,
user_id=user.id,
amount="80.00", currency="EUR",
description=f"Admin debt {uuid4().hex[:6]}",
revenue_account=standard_accounts["revenue_rent"]["name"],
)
expense = await post_expense(
client,
wallet_inkey=wallet.inkey,
user_wallet_id=wallet.id,
amount="30.00", currency="EUR",
description=f"User expense {uuid4().hex[:6]}",
expense_account=standard_accounts["expense_food"]["name"],
)
await _approve_and_refresh(client, wallet, super_user_headers, expense["id"])
balance = await get_balance(client, wallet_inkey=wallet.inkey)
eur = balance.get("fiat_balances", {}).get("EUR")
assert float(eur) == pytest.approx(50.0), (
f"expected +50 EUR (80 - 30), got {eur} from {balance}"
)
# ---------------------------------------------------------------------------
# Lifetime totals (separate from net balance)
# ---------------------------------------------------------------------------
@pytest.mark.anyio
async def test_lifetime_totals_track_originals_not_net(
client, super_user_headers, configured_user, standard_accounts,
):
"""`total_expenses_fiat` and `total_income_fiat` track originally-entered
amounts, not net obligation — see the `models.py:93` invariant. Even
after partial-direction submissions, the totals should equal the gross.
"""
_, wallet = configured_user
expense = await post_expense(
client,
wallet_inkey=wallet.inkey,
user_wallet_id=wallet.id,
amount="45.00", currency="EUR",
description=f"e1 {uuid4().hex[:6]}",
expense_account=standard_accounts["expense_food"]["name"],
)
income = await post_income(
client,
wallet_inkey=wallet.inkey,
amount="80.00", currency="EUR",
description=f"i1 {uuid4().hex[:6]}",
revenue_account=standard_accounts["revenue_rent"]["name"],
)
await _approve_and_refresh(client, wallet, super_user_headers, expense["id"])
await _approve_and_refresh(client, wallet, super_user_headers, income["id"])
balance = await get_balance(client, wallet_inkey=wallet.inkey)
exp_eur = balance.get("total_expenses_fiat", {}).get("EUR", 0)
inc_eur = balance.get("total_income_fiat", {}).get("EUR", 0)
assert float(exp_eur) == pytest.approx(45.0), (
f"total_expenses_fiat should be gross 45, got {exp_eur}"
)
assert float(inc_eur) == pytest.approx(80.0), (
f"total_income_fiat should be gross 80, got {inc_eur}"
)
# ---------------------------------------------------------------------------
# Exclusions — pending and voided
# ---------------------------------------------------------------------------
@pytest.mark.anyio
async def test_pending_entries_excluded_from_balance(
client, super_user_headers, configured_user, standard_accounts,
):
"""Two expenses submitted, only one approved → only the approved one
moves the balance."""
_, wallet = configured_user
approved = await post_expense(
client,
wallet_inkey=wallet.inkey,
user_wallet_id=wallet.id,
amount="25.00", currency="EUR",
description=f"approved-only {uuid4().hex[:6]}",
expense_account=standard_accounts["expense_food"]["name"],
)
# Submit a second expense but leave it pending.
await post_expense(
client,
wallet_inkey=wallet.inkey,
user_wallet_id=wallet.id,
amount="1000.00", currency="EUR",
description=f"pending-not-counted {uuid4().hex[:6]}",
expense_account=standard_accounts["expense_supplies"]["name"],
)
await _approve_and_refresh(client, wallet, super_user_headers, approved["id"])
balance = await get_balance(client, wallet_inkey=wallet.inkey)
eur = balance.get("fiat_balances", {}).get("EUR")
assert float(eur) == pytest.approx(-25.0), (
f"only approved expense should count; pending 1000 must be excluded. "
f"got {eur}"
)
@pytest.mark.anyio
async def test_voided_entries_excluded_from_balance(
client, super_user_headers, configured_user, standard_accounts,
):
"""A voided entry stops contributing to the balance the moment it's
rejected — verified by submitting then rejecting and confirming the
balance is what it would be without that entry."""
_, wallet = configured_user
keep = await post_expense(
client,
wallet_inkey=wallet.inkey,
user_wallet_id=wallet.id,
amount="35.00", currency="EUR",
description=f"keep {uuid4().hex[:6]}",
expense_account=standard_accounts["expense_food"]["name"],
)
rejected = await post_expense(
client,
wallet_inkey=wallet.inkey,
user_wallet_id=wallet.id,
amount="500.00", currency="EUR",
description=f"will-be-voided {uuid4().hex[:6]}",
expense_account=standard_accounts["expense_food"]["name"],
)
await _approve_and_refresh(client, wallet, super_user_headers, keep["id"])
await reject_entry(
client, super_user_headers=super_user_headers, entry_id=rejected["id"],
)
balance = await get_balance(client, wallet_inkey=wallet.inkey)
eur = balance.get("fiat_balances", {}).get("EUR")
assert float(eur) == pytest.approx(-35.0), (
f"voided 500 must not contribute; only the 35 EUR keeper. got {eur}"
)
# ---------------------------------------------------------------------------
# Admin /balances/all
# ---------------------------------------------------------------------------
@pytest.mark.anyio
async def test_admin_balances_all_includes_users_with_obligations(
client, super_user_headers, configured_user, configured_user_b,
standard_accounts,
):
"""`/balances/all` returns one row per user that has any Payable or
Receivable activity. Two users → two rows after both submit + approve.
"""
user_a, wallet_a = configured_user
user_b, wallet_b = configured_user_b
a_entry = await post_expense(
client,
wallet_inkey=wallet_a.inkey,
user_wallet_id=wallet_a.id,
amount="60.00", currency="EUR",
description=f"A-bal {uuid4().hex[:6]}",
expense_account=standard_accounts["expense_food"]["name"],
)
b_entry = await post_expense(
client,
wallet_inkey=wallet_b.inkey,
user_wallet_id=wallet_b.id,
amount="90.00", currency="EUR",
description=f"B-bal {uuid4().hex[:6]}",
expense_account=standard_accounts["expense_food"]["name"],
)
await _approve_and_refresh(client, wallet_a, super_user_headers, a_entry["id"])
await _approve_and_refresh(client, wallet_b, super_user_headers, b_entry["id"])
rows = await get_all_balances(client, super_user_headers=super_user_headers)
by_id = {r.get("user_id")[:8]: r for r in rows if r.get("user_id")}
assert user_a.id[:8] in by_id, f"user A missing from /balances/all"
assert user_b.id[:8] in by_id, f"user B missing from /balances/all"
a_eur = by_id[user_a.id[:8]].get("fiat_balances", {}).get("EUR")
b_eur = by_id[user_b.id[:8]].get("fiat_balances", {}).get("EUR")
assert float(a_eur) == pytest.approx(-60.0), (
f"user A EUR balance wrong in /balances/all: {a_eur}"
)
assert float(b_eur) == pytest.approx(-90.0), (
f"user B EUR balance wrong in /balances/all: {b_eur}"
)
@pytest.mark.anyio
async def test_non_super_user_cannot_get_all_balances(
client, configured_user,
):
"""`/balances/all` is admin-only — regular user wallet admin-key 403s."""
_, wallet = configured_user
r = await client.get(
"/libra/api/v1/balances/all",
headers={"X-Api-Key": wallet.adminkey, "Content-type": "application/json"},
)
assert r.status_code == 403, f"expected 403, got {r.status_code}: {r.text}"
assert "super" in r.text.lower()
# ---------------------------------------------------------------------------
# Credit balance — libra-#41
# ---------------------------------------------------------------------------
@pytest.mark.anyio
async def test_credit_balance_subtracts_from_net(
client, configured_user,
):
"""A user-credit balance on `Liabilities:Credit:User-X` flows into the
displayed net so the user-facing balance is always honest about what
libra owes them.
`#41` will land the settlement-side overflow logic that writes credit
automatically. This test pre-creates the credit account and posts a
balanced credit-bearing transaction directly via Fava so we can lock
in the BQL-side behaviour (`get_user_balance_bql` includes the Credit
namespace alongside Payable + Receivable) ahead of the settlement
endpoint changes in #14.
"""
user, wallet = configured_user
fava_client_mod = _libra_module("fava_client")
fava = fava_client_mod.get_fava_client()
# Open the per-user credit account in Beancount. The settlement endpoint
# will do this via `get_or_create_user_account` when #14 lands.
credit_account = f"Liabilities:Credit:User-{user.id[:8]}"
await fava.add_account(credit_account, currencies=["EUR", "SATS"])
# Manually post a balanced entry mimicking what the future settlement
# overflow leg looks like in isolation:
# DR Assets:Cash +30 EUR (libra receives cash)
# CR Liabilities:Credit -30 EUR (libra owes user that 30 going forward)
tag = uuid4().hex[:6]
beancount_format = _libra_module("beancount_format")
entry = beancount_format.format_transaction(
date_val=date.today(),
flag="*",
narration=f"Credit-balance test {tag}",
postings=[
{"account": "Assets:Cash", "amount": "30.00 EUR"},
{"account": credit_account, "amount": "-30.00 EUR"},
],
tags=["credit-test"],
links=[f"credit-test-{tag}"],
meta={"user-id": user.id, "source": "test"},
)
await fava.add_entry(entry)
# Force a fresh Fava read before the BQL balance query (libra-#37).
await list_user_entries(client, wallet_inkey=wallet.inkey)
# The user's EUR balance should now read -30 (libra owes user 30 via
# credit). Without the BQL change, this would read 0 because the query
# would skip the Credit namespace entirely.
balance = await get_balance(client, wallet_inkey=wallet.inkey)
eur = balance.get("fiat_balances", {}).get("EUR")
assert eur is not None, f"missing EUR in fiat_balances: {balance}"
assert float(eur) == pytest.approx(-30.0), (
f"expected -30 EUR (libra owes user via credit), got {eur} from {balance}"
)
# The accounts breakdown should surface the credit row so UIs can render
# it as a distinct line item per #41's display contract. `accounts` (the
# legacy field on UserBalance) stays empty for back-compat; the new
# `account_balances` field carries the BQL per-account breakdown.
account_balances = balance.get("account_balances", [])
credit_rows = [
a for a in account_balances if "Credit" in (a.get("account") or "")
]
assert credit_rows, (
f"credit account missing from breakdown — UI can't render 'You have "
f"30 EUR credit' line item. account_balances: {account_balances}"
)

219
tests/test_entries_admin_api.py
View file

@ -1,219 +0,0 @@
"""Admin-side journal entry endpoints — receivable and revenue.
- `POST /libra/api/v1/entries/receivable` — admin records that a user owes
libra. Lands as a pending (`!`) entry, balance untouched until approve.
- `POST /libra/api/v1/entries/revenue` — admin records that libra received
a payment unrelated to any user. Lands as a cleared (`*`) entry, no
approval needed.
Auth gate covered too: a regular user's wallet admin-key passes
`require_admin_key` but fails the super-user identity check in libra's own
`require_super_user`, so the endpoint returns 403.
"""
from uuid import uuid4
import pytest
from .helpers import (
get_balance,
list_user_entries,
post_receivable,
post_revenue,
)
@pytest.mark.anyio
async def test_admin_records_receivable_lands_cleared(
client, super_user_headers, configured_user, standard_accounts,
):
"""Admin posts a receivable for a user — the Beancount entry is written
with the cleared `*` flag immediately (not pending). The user's balance
reflects the debt without an approve step.
Note: `JournalEntry.flag` in the API response is misleading — it's a
leftover of the legacy model and reports PENDING, but the entry in
Beancount is written as `*`. The on-disk reality is what affects the
balance, so that's what we assert.
"""
user, wallet = configured_user
response = await post_receivable(
client,
super_user_headers=super_user_headers,
user_id=user.id,
amount="200.00",
currency="EUR",
description=f"December rent share {uuid4().hex[:6]}",
revenue_account=standard_accounts["revenue_rent"]["name"],
)
assert response.get("id"), f"expected id in response, got {response}"
# Force a fresh Fava read before checking balance — Fava lazily reloads
# the .beancount file and a balance call right after add_entry can hit
# a stale view.
await list_user_entries(client, wallet_inkey=wallet.inkey)
balance = await get_balance(client, wallet_inkey=wallet.inkey)
eur = balance.get("fiat_balances", {}).get("EUR")
assert eur is not None, f"expected EUR in fiat_balances, got {balance}"
assert float(eur) == pytest.approx(200.0), (
f"expected +200 EUR (user-owes-libra) after receivable, got {eur}"
)
@pytest.mark.anyio
async def test_receivable_visible_in_target_users_journal(
client, super_user_headers, configured_user, standard_accounts,
):
"""The receivable shows up in the *debtor* user's journal listing
(not just in the admin view)."""
user, wallet = configured_user
tag = uuid4().hex[:6]
await post_receivable(
client,
super_user_headers=super_user_headers,
user_id=user.id,
amount="75.00",
currency="EUR",
description=f"Workshop fee {tag}",
revenue_account=standard_accounts["revenue_fees"]["name"],
)
listing = await list_user_entries(client, wallet_inkey=wallet.inkey)
descriptions = [e.get("description") or "" for e in listing.get("entries", [])]
assert any(tag in d for d in descriptions), (
f"receivable missing from debtor's journal: {descriptions}"
)
@pytest.mark.anyio
async def test_admin_records_revenue_clears_immediately(
client, super_user_headers, standard_accounts,
):
"""Revenue (libra received money, no user debt) is cleared on creation —
no admin approval step."""
response = await post_revenue(
client,
super_user_headers=super_user_headers,
amount="500.00",
currency="EUR",
description=f"Workshop fees collected {uuid4().hex[:6]}",
revenue_account=standard_accounts["revenue_fees"]["name"],
payment_method_account="Assets:Cash",
)
assert response.get("id"), f"expected id in response, got {response}"
# Cleared on creation — flag is `*`, no approve_entry call needed.
@pytest.mark.anyio
async def test_non_super_user_cannot_post_receivable(
client, configured_user, standard_accounts,
):
"""A regular user's wallet admin key passes `require_admin_key` but
fails libra's super-user identity check. Returns 403."""
user, wallet = configured_user
admin_key_headers = {"X-Api-Key": wallet.adminkey, "Content-type": "application/json"}
r = await client.post(
"/libra/api/v1/entries/receivable",
headers=admin_key_headers,
json={
"user_id": user.id,
"amount": "10.00",
"currency": "EUR",
"description": "Should be denied",
"revenue_account": standard_accounts["revenue_rent"]["name"],
},
)
assert r.status_code == 403, f"expected 403, got {r.status_code}: {r.text}"
assert "super" in r.text.lower(), (
f"expected super-user error message, got {r.text!r}"
)
@pytest.mark.anyio
async def test_non_super_user_cannot_post_revenue(
client, configured_user, standard_accounts,
):
"""Same super-user gate covers the revenue endpoint."""
_, wallet = configured_user
admin_key_headers = {"X-Api-Key": wallet.adminkey, "Content-type": "application/json"}
r = await client.post(
"/libra/api/v1/entries/revenue",
headers=admin_key_headers,
json={
"amount": "10.00",
"currency": "EUR",
"description": "Should be denied",
"revenue_account": standard_accounts["revenue_fees"]["name"],
"payment_method_account": "Assets:Cash",
},
)
assert r.status_code == 403, f"expected 403, got {r.status_code}: {r.text}"
assert "super" in r.text.lower()
@pytest.mark.anyio
async def test_receivable_unknown_revenue_account_returns_404(
client, super_user_headers, configured_user,
):
"""An admin posting against a non-existent revenue account gets 404."""
user, _ = configured_user
r = await client.post(
"/libra/api/v1/entries/receivable",
headers=super_user_headers,
json={
"user_id": user.id,
"amount": "10.00",
"currency": "EUR",
"description": "Bad account",
"revenue_account": f"Income:Test:DoesNotExist-{uuid4().hex[:6]}",
},
)
assert r.status_code == 404, f"expected 404, got {r.status_code}: {r.text}"
assert "not found" in r.text.lower()
@pytest.mark.anyio
async def test_receivable_unknown_currency_returns_400(
client, super_user_headers, configured_user, standard_accounts,
):
"""Currency validation hits before account lookups."""
user, _ = configured_user
r = await client.post(
"/libra/api/v1/entries/receivable",
headers=super_user_headers,
json={
"user_id": user.id,
"amount": "10.00",
"currency": "XYZ",
"description": "Bogus currency",
"revenue_account": standard_accounts["revenue_rent"]["name"],
},
)
assert r.status_code == 400, f"expected 400, got {r.status_code}: {r.text}"
assert "currency" in r.text.lower() or "xyz" in r.text.lower()
@pytest.mark.anyio
async def test_revenue_unknown_payment_account_returns_404(
client, super_user_headers, standard_accounts,
):
"""Revenue endpoint validates BOTH accounts; the payment-method one too."""
r = await client.post(
"/libra/api/v1/entries/revenue",
headers=super_user_headers,
json={
"amount": "10.00",
"currency": "EUR",
"description": "Bad payment account",
"revenue_account": standard_accounts["revenue_fees"]["name"],
"payment_method_account": f"Assets:DoesNotExist-{uuid4().hex[:6]}",
},
)
assert r.status_code == 404, f"expected 404, got {r.status_code}: {r.text}"
assert "not found" in r.text.lower()

211
tests/test_entries_user_api.py
View file

@ -1,211 +0,0 @@
"""User-side expense submission flow — `POST /libra/api/v1/entries/expense`.
Covers:
- Submission lands as a pending entry, visible to the user, doesn't move
the cleared-only balance.
- Cross-user isolation — user B can't see user A's entries.
- Permission gating, currency validation, missing user-wallet setup.
- Multiple submissions accumulate in the user journal listing.
Settlement, approval, and balance-after-approval are exercised in
`test_smoke.py` (one canonical path) and `test_balances_api.py` (the mixed
income+expense display scenario the user named).
"""
from uuid import uuid4
import pytest
from .helpers import (
create_account,
get_balance,
list_user_entries,
post_expense,
)
@pytest.mark.anyio
async def test_expense_creates_pending_entry_visible_in_user_journal(
client, configured_user, standard_accounts,
):
"""Submitting an expense creates a pending (`!`) entry the user can see
immediately. The cleared-only balance query is unchanged because pending
entries are excluded."""
_, wallet = configured_user
response = await post_expense(
client,
wallet_inkey=wallet.inkey,
user_wallet_id=wallet.id,
amount="25.00",
currency="EUR",
description="Test groceries",
expense_account=standard_accounts["expense_food"]["name"],
)
assert response.get("id"), f"expected id in response, got {response}"
listing = await list_user_entries(client, wallet_inkey=wallet.inkey)
entries = listing.get("entries", [])
assert any(
"Test groceries" in (e.get("description") or "") for e in entries
), f"submitted expense missing from /entries/user: {entries}"
bal = await get_balance(client, wallet_inkey=wallet.inkey)
assert not bal.get("fiat_balances"), (
f"pending entry should not affect cleared balance, got {bal}"
)
@pytest.mark.anyio
async def test_user_cannot_see_other_users_entries(
client, configured_user, configured_user_b, standard_accounts,
):
"""User A submits an expense; user B's `/entries/user` listing is
scoped to B and never references A's user-id account fragment."""
user_a, wallet_a = configured_user
_, wallet_b = configured_user_b
await post_expense(
client,
wallet_inkey=wallet_a.inkey,
user_wallet_id=wallet_a.id,
amount="40.00",
currency="EUR",
description=f"A-private-{uuid4().hex[:6]}",
expense_account=standard_accounts["expense_food"]["name"],
)
listing_b = await list_user_entries(client, wallet_inkey=wallet_b.inkey)
a_short = user_a.id[:8]
for entry in listing_b.get("entries", []):
for posting in entry.get("postings", []):
assert a_short not in posting.get("account", ""), (
f"user B's listing leaked user A's account: {posting}"
)
@pytest.mark.anyio
async def test_expense_without_permission_returns_403(
client, super_user_headers, configured_user,
):
"""Submitting to an expense account the user has no `submit_expense`
permission on returns 403 with a permission-error detail."""
_, wallet = configured_user
# Fresh expense account that no permission was granted on.
new_account = await create_account(
client,
super_user_headers=super_user_headers,
name=f"Expenses:Test:Unguarded-{uuid4().hex[:6]}",
account_type="expense",
)
r = await client.post(
"/libra/api/v1/entries/expense",
headers={"X-Api-Key": wallet.inkey, "Content-type": "application/json"},
json={
"description": "Should be denied",
"amount": "10.00",
"currency": "EUR",
"expense_account": new_account["name"],
"user_wallet": wallet.id,
"is_equity": False,
},
)
assert r.status_code == 403, f"expected 403, got {r.status_code}: {r.text}"
assert "permission" in r.text.lower(), (
f"expected permission error message, got {r.text!r}"
)
@pytest.mark.anyio
async def test_expense_with_unknown_currency_returns_400(
client, configured_user, standard_accounts,
):
"""An unsupported currency is rejected with 400 before any Fava call."""
_, wallet = configured_user
r = await client.post(
"/libra/api/v1/entries/expense",
headers={"X-Api-Key": wallet.inkey, "Content-type": "application/json"},
json={
"description": "Unknown currency",
"amount": "10.00",
"currency": "XYZ",
"expense_account": standard_accounts["expense_food"]["name"],
"user_wallet": wallet.id,
"is_equity": False,
},
)
assert r.status_code == 400, f"expected 400, got {r.status_code}: {r.text}"
assert "currency" in r.text.lower(), (
f"expected currency error message, got {r.text!r}"
)
@pytest.mark.anyio
async def test_expense_without_user_wallet_configured_returns_400(
client, libra_user, libra_wallet, standard_accounts, # noqa: ARG001 (libra_wallet ensures session-level setup)
):
"""A user whose own libra wallet isn't configured can't submit expenses.
`libra_user` (vs `configured_user`) skips the `PUT /user/wallet` step
on purpose so the precondition fires.
"""
_, wallet = libra_user
r = await client.post(
"/libra/api/v1/entries/expense",
headers={"X-Api-Key": wallet.inkey, "Content-type": "application/json"},
json={
"description": "Missing user wallet setup",
"amount": "10.00",
"currency": "EUR",
"expense_account": standard_accounts["expense_food"]["name"],
"user_wallet": wallet.id,
"is_equity": False,
},
)
assert r.status_code == 400, f"expected 400, got {r.status_code}: {r.text}"
assert "wallet" in r.text.lower(), (
f"expected wallet-config error, got {r.text!r}"
)
@pytest.mark.anyio
async def test_multiple_expenses_accumulate_in_user_journal(
client, configured_user, standard_accounts,
):
"""Each submission shows up in `/entries/user`; the listing's `total`
grows by exactly the number of submissions."""
_, wallet = configured_user
initial = await list_user_entries(client, wallet_inkey=wallet.inkey)
initial_total = initial.get("total", 0)
tag = uuid4().hex[:6]
descriptions = [f"Coffee-{tag}", f"Bread-{tag}", f"Vegetables-{tag}"]
for description in descriptions:
await post_expense(
client,
wallet_inkey=wallet.inkey,
user_wallet_id=wallet.id,
amount="7.50",
currency="EUR",
description=description,
expense_account=standard_accounts["expense_food"]["name"],
)
final = await list_user_entries(client, wallet_inkey=wallet.inkey)
final_total = final.get("total", 0)
assert final_total - initial_total == len(descriptions), (
f"expected total to grow by {len(descriptions)}, "
f"went from {initial_total} to {final_total}"
)
# Libra appends " (<amount> <currency>)" to entry descriptions, so check
# substring rather than exact match.
final_descs = [e.get("description") or "" for e in final.get("entries", [])]
for description in descriptions:
assert any(description in d for d in final_descs), (
f"missing {description} from journal listing: {final_descs}"
)

205
tests/test_lightning_api.py
View file

@ -1,205 +0,0 @@
"""Lightning payment flow — `POST /generate-payment-invoice` and
`POST /record-payment`.
- User has a balance owed to libra → user generates an invoice on the libra
wallet → user pays it → `/record-payment` records the settlement entry.
## Coverage status
This file covers auth gates and error paths that don't require an active
Lightning backend. Tests that actually need invoice generation are skipped
because:
- The default `VoidWallet` 500s on any invoice operation.
- Switching to `FakeWallet` (via `settings.lnbits_backend_wallet_class`)
DOES enable invoice generation, but the LifespanManager teardown then
hangs indefinitely under anyio's TestRunner — some Lightning-side
background task doesn't unwind cleanly. Investigation deferred; the
auth gates + 404/400 error paths are what we can lock in for now.
The skipped tests carry full implementations so flipping them back on is
a one-line change once the teardown issue is resolved (or once we move to
a subprocess-based runner for the LN file).
"""
from uuid import uuid4
import pytest
from .helpers import (
list_user_entries,
post_receivable,
)
NEEDS_LIGHTNING_BACKEND = pytest.mark.skip(
reason="Tracked by libra/issues/40 — VoidWallet 500s, FakeWallet hangs the "
"LifespanManager teardown under anyio's TestRunner. Flip when resolved."
)
async def _setup_receivable_balance(
client, super_user_headers, configured_user, standard_accounts,
amount="100.00",
):
"""Helper: create + (auto-cleared) receivable so the user has a balance
owed to libra. Returns the (user, wallet) pair."""
user, wallet = configured_user
await post_receivable(
client,
super_user_headers=super_user_headers,
user_id=user.id,
amount=amount, currency="EUR",
description=f"Setup debt {uuid4().hex[:6]}",
revenue_account=standard_accounts["revenue_rent"]["name"],
)
# Force a Fava reload before downstream BQL balance reads (see #37).
await list_user_entries(client, wallet_inkey=wallet.inkey)
return user, wallet
# ---------------------------------------------------------------------------
# /generate-payment-invoice
# ---------------------------------------------------------------------------
@NEEDS_LIGHTNING_BACKEND
@pytest.mark.anyio
async def test_user_can_generate_invoice_for_own_balance(
client, super_user_headers, configured_user, standard_accounts,
):
"""User with a receivable generates an invoice on the libra wallet.
Response carries the bolt11 string and the libra wallet's inkey for
the client to poll payment status."""
_, wallet = await _setup_receivable_balance(
client, super_user_headers, configured_user, standard_accounts,
)
r = await client.post(
"/libra/api/v1/generate-payment-invoice",
headers={"X-Api-Key": wallet.inkey, "Content-type": "application/json"},
json={"amount": 50_000}, # 50k sats partial settlement
)
assert r.status_code == 200, f"generate-invoice: {r.status_code} {r.text}"
payload = r.json()
assert payload.get("payment_hash"), f"missing payment_hash: {payload}"
assert payload.get("payment_request"), f"missing bolt11 payment_request: {payload}"
assert payload.get("amount") == 50_000
assert payload.get("check_wallet_key"), f"missing check_wallet_key: {payload}"
@NEEDS_LIGHTNING_BACKEND
@pytest.mark.anyio
async def test_super_user_can_generate_invoice_for_another_user(
client, super_user_headers, libra_wallet, configured_user, standard_accounts,
):
"""Admin generating an invoice on behalf of a user — uses the libra
wallet's admin key + body `user_id`. The endpoint actually requires
`wallet.wallet.user == super_user` (which is the libra wallet owner).
Generate-invoice is `require_invoice_key`-gated so we pass the libra
wallet's invoice key, and the user_id field opts into "for that user".
"""
user, _ = await _setup_receivable_balance(
client, super_user_headers, configured_user, standard_accounts,
)
r = await client.post(
"/libra/api/v1/generate-payment-invoice",
headers={"X-Api-Key": libra_wallet.inkey, "Content-type": "application/json"},
json={"amount": 30_000, "user_id": user.id},
)
assert r.status_code == 200, f"admin generate-invoice: {r.status_code} {r.text}"
assert r.json().get("payment_request"), "admin-generated invoice missing bolt11"
@pytest.mark.anyio
async def test_non_super_user_cannot_generate_invoice_for_another_user(
client, super_user_headers, configured_user, configured_user_b,
standard_accounts,
):
"""A regular user cannot pass `user_id` and have libra generate an
invoice on someone else's behalf — 403."""
user_a, _ = await _setup_receivable_balance(
client, super_user_headers, configured_user, standard_accounts,
)
_, wallet_b = configured_user_b
r = await client.post(
"/libra/api/v1/generate-payment-invoice",
headers={"X-Api-Key": wallet_b.inkey, "Content-type": "application/json"},
json={"amount": 10_000, "user_id": user_a.id},
)
assert r.status_code == 403, f"expected 403, got {r.status_code}: {r.text}"
@pytest.mark.anyio
async def test_generate_invoice_without_auth_returns_401(client):
"""Invoice-key auth required — no header → 401."""
r = await client.post(
"/libra/api/v1/generate-payment-invoice",
json={"amount": 10_000},
)
assert r.status_code == 401, f"expected 401, got {r.status_code}: {r.text}"
# ---------------------------------------------------------------------------
# /record-payment
# ---------------------------------------------------------------------------
@pytest.mark.anyio
async def test_record_payment_unknown_hash_returns_404(
client, configured_user,
):
"""Recording a payment hash that doesn't correspond to a real payment
in LNbits returns 404."""
_, wallet = configured_user
r = await client.post(
"/libra/api/v1/record-payment",
headers={"X-Api-Key": wallet.inkey, "Content-type": "application/json"},
json={"payment_hash": "0" * 64},
)
assert r.status_code == 404, f"expected 404, got {r.status_code}: {r.text}"
assert "payment not found" in r.text.lower() or "payment" in r.text.lower()
@NEEDS_LIGHTNING_BACKEND
@pytest.mark.anyio
async def test_record_payment_pending_invoice_returns_400(
client, super_user_headers, configured_user, standard_accounts,
):
"""A freshly-generated invoice that hasn't been paid yet is pending —
`/record-payment` must reject it with 400 rather than silently
recording a non-existent settlement."""
_, wallet = await _setup_receivable_balance(
client, super_user_headers, configured_user, standard_accounts,
)
# Generate an invoice on the libra wallet.
gen = await client.post(
"/libra/api/v1/generate-payment-invoice",
headers={"X-Api-Key": wallet.inkey, "Content-type": "application/json"},
json={"amount": 15_000},
)
assert gen.status_code == 200
payment_hash = gen.json()["payment_hash"]
# Try to record it before any payment lands.
r = await client.post(
"/libra/api/v1/record-payment",
headers={"X-Api-Key": wallet.inkey, "Content-type": "application/json"},
json={"payment_hash": payment_hash},
)
assert r.status_code == 400, f"expected 400, got {r.status_code}: {r.text}"
assert "not yet settled" in r.text.lower() or "pending" in r.text.lower(), (
f"expected pending/settled message, got {r.text!r}"
)
@pytest.mark.anyio
async def test_record_payment_without_auth_returns_401(client):
r = await client.post(
"/libra/api/v1/record-payment",
json={"payment_hash": "abc"},
)
assert r.status_code == 401, f"expected 401, got {r.status_code}: {r.text}"

307
tests/test_manual_payment_requests_api.py
View file

@ -1,307 +0,0 @@
"""Manual payment request flow — user asks for libra to pay them via a
non-Lightning route (cash, bank, etc.); admin approves or rejects.
Endpoints:
- `POST /libra/api/v1/manual-payment-request` (invoice key, user)
- `GET /libra/api/v1/manual-payment-requests` (invoice key, own only)
- `GET /libra/api/v1/manual-payment-requests/all` (super user, all)
- `POST /libra/api/v1/manual-payment-requests/{id}/approve` (super user)
- `POST /libra/api/v1/manual-payment-requests/{id}/reject` (super user)
The amount in the request body is in **satoshis** (no fiat conversion at this
endpoint — `CreateManualPaymentRequest` has `amount: int`).
Approve creates a Beancount payment entry:
DR Liabilities:Payable:User-{id} (zeroes libra's debt to the user)
CR Assets:Bitcoin:Lightning (cash leaves libra)
"""
from uuid import uuid4
import pytest
from .helpers import (
approve_manual_payment_request,
reject_manual_payment_request,
submit_manual_payment_request,
)
# ---------------------------------------------------------------------------
# User-side submission
# ---------------------------------------------------------------------------
@pytest.mark.anyio
async def test_user_can_submit_manual_payment_request(
client, configured_user,
):
"""Submission returns 200 with a pending request and the user's id."""
user, wallet = configured_user
desc = f"Coffee reimbursement {uuid4().hex[:6]}"
result = await submit_manual_payment_request(
client,
wallet_inkey=wallet.inkey,
amount_sats=50_000,
description=desc,
)
assert result.get("id"), f"missing id: {result}"
assert result.get("user_id") == user.id
assert result.get("amount") == 50_000
assert result.get("description") == desc
assert result.get("status") == "pending"
@pytest.mark.anyio
async def test_user_lists_own_manual_payment_requests(
client, configured_user,
):
"""The user-side listing returns the requests this user submitted."""
_, wallet = configured_user
tag = uuid4().hex[:6]
submitted = await submit_manual_payment_request(
client,
wallet_inkey=wallet.inkey,
amount_sats=12_000,
description=f"list-test {tag}",
)
r = await client.get(
"/libra/api/v1/manual-payment-requests",
headers={"X-Api-Key": wallet.inkey},
)
assert r.status_code == 200, f"list: {r.status_code} {r.text}"
ids = [req.get("id") for req in r.json()]
assert submitted["id"] in ids, f"submitted request missing from listing: {ids}"
@pytest.mark.anyio
async def test_user_cannot_see_another_users_manual_payment_requests(
client, configured_user, configured_user_b,
):
"""User-side listing is scoped to the calling user, not all requests."""
user_a, wallet_a = configured_user
_, wallet_b = configured_user_b
submitted_a = await submit_manual_payment_request(
client,
wallet_inkey=wallet_a.inkey,
amount_sats=8_000,
description=f"A-private {uuid4().hex[:6]}",
)
r = await client.get(
"/libra/api/v1/manual-payment-requests",
headers={"X-Api-Key": wallet_b.inkey},
)
assert r.status_code == 200
user_ids = {req.get("user_id") for req in r.json()}
ids = [req.get("id") for req in r.json()]
assert submitted_a["id"] not in ids, (
f"user B saw user A's request: {submitted_a['id']} in {ids}"
)
assert user_a.id not in user_ids, (
f"user B's listing contained user A's id: {user_ids}"
)
# ---------------------------------------------------------------------------
# Admin listing
# ---------------------------------------------------------------------------
@pytest.mark.anyio
async def test_admin_can_list_all_manual_payment_requests(
client, super_user_headers, configured_user, configured_user_b,
):
"""The admin listing returns requests from any user."""
_, wallet_a = configured_user
_, wallet_b = configured_user_b
a_req = await submit_manual_payment_request(
client,
wallet_inkey=wallet_a.inkey,
amount_sats=10_000,
description=f"A {uuid4().hex[:6]}",
)
b_req = await submit_manual_payment_request(
client,
wallet_inkey=wallet_b.inkey,
amount_sats=20_000,
description=f"B {uuid4().hex[:6]}",
)
r = await client.get(
"/libra/api/v1/manual-payment-requests/all",
headers=super_user_headers,
)
assert r.status_code == 200, f"admin list: {r.status_code} {r.text}"
ids = [req.get("id") for req in r.json()]
assert a_req["id"] in ids and b_req["id"] in ids, (
f"admin list missing entries: ids={ids}"
)
@pytest.mark.anyio
async def test_admin_listing_status_filter(
client, super_user_headers, configured_user,
):
"""`?status=pending` returns only the pending requests."""
_, wallet = configured_user
submitted = await submit_manual_payment_request(
client,
wallet_inkey=wallet.inkey,
amount_sats=5_000,
description=f"pending-filter {uuid4().hex[:6]}",
)
r = await client.get(
"/libra/api/v1/manual-payment-requests/all?status=pending",
headers=super_user_headers,
)
assert r.status_code == 200, f"filtered list: {r.status_code} {r.text}"
statuses = {req.get("status") for req in r.json()}
assert statuses == {"pending"}, f"non-pending rows in filtered list: {statuses}"
assert submitted["id"] in [req.get("id") for req in r.json()]
@pytest.mark.anyio
async def test_non_super_user_cannot_list_all_requests(
client, configured_user,
):
"""Wallet admin-key of a non-super user fails the super-user check."""
_, wallet = configured_user
r = await client.get(
"/libra/api/v1/manual-payment-requests/all",
headers={"X-Api-Key": wallet.adminkey, "Content-type": "application/json"},
)
assert r.status_code == 403, f"expected 403, got {r.status_code}: {r.text}"
assert "super" in r.text.lower()
# ---------------------------------------------------------------------------
# Approve / reject
# ---------------------------------------------------------------------------
@pytest.mark.anyio
async def test_admin_can_reject_manual_payment_request(
client, super_user_headers, configured_user,
):
"""Reject flips status to 'rejected' and doesn't touch Beancount."""
_, wallet = configured_user
submitted = await submit_manual_payment_request(
client,
wallet_inkey=wallet.inkey,
amount_sats=3_500,
description=f"reject me {uuid4().hex[:6]}",
)
result = await reject_manual_payment_request(
client, super_user_headers=super_user_headers, request_id=submitted["id"],
)
assert result.get("status") == "rejected"
@pytest.mark.anyio
async def test_rejecting_already_rejected_returns_400(
client, super_user_headers, configured_user,
):
"""The endpoint guards against double-decisions."""
_, wallet = configured_user
submitted = await submit_manual_payment_request(
client,
wallet_inkey=wallet.inkey,
amount_sats=4_000,
description=f"double reject {uuid4().hex[:6]}",
)
await reject_manual_payment_request(
client, super_user_headers=super_user_headers, request_id=submitted["id"],
)
r = await client.post(
f"/libra/api/v1/manual-payment-requests/{submitted['id']}/reject",
headers=super_user_headers,
)
assert r.status_code == 400, f"expected 400, got {r.status_code}: {r.text}"
assert "reject" in r.text.lower()
@pytest.mark.anyio
async def test_approve_unknown_request_returns_404(
client, super_user_headers,
):
r = await client.post(
f"/libra/api/v1/manual-payment-requests/{uuid4().hex[:16]}/approve",
headers=super_user_headers,
)
assert r.status_code == 404, f"expected 404, got {r.status_code}: {r.text}"
@pytest.mark.anyio
async def test_non_super_user_cannot_approve(
client, configured_user,
):
_, wallet = configured_user
submitted = await submit_manual_payment_request(
client,
wallet_inkey=wallet.inkey,
amount_sats=2_000,
description=f"no approve for you {uuid4().hex[:6]}",
)
r = await client.post(
f"/libra/api/v1/manual-payment-requests/{submitted['id']}/approve",
headers={"X-Api-Key": wallet.adminkey, "Content-type": "application/json"},
)
assert r.status_code == 403, f"expected 403, got {r.status_code}: {r.text}"
@pytest.mark.anyio
async def test_admin_can_approve_manual_payment_request(
client, super_user_headers, configured_user, standard_accounts,
# noqa: ARG001 (standard_accounts ensures Assets:Bitcoin:Lightning exists)
):
"""Approve creates a Beancount payment entry and flips status to
'approved'. Requires `Assets:Bitcoin:Lightning` to exist in libra's
local DB (provided by the `standard_accounts` fixture)."""
_, wallet = configured_user
submitted = await submit_manual_payment_request(
client,
wallet_inkey=wallet.inkey,
amount_sats=6_000,
description=f"approve me {uuid4().hex[:6]}",
)
result = await approve_manual_payment_request(
client, super_user_headers=super_user_headers, request_id=submitted["id"],
)
assert result.get("status") == "approved"
assert result.get("id") == submitted["id"]
@pytest.mark.anyio
async def test_approving_already_approved_returns_400(
client, super_user_headers, configured_user, standard_accounts,
):
"""Idempotency guard: second approve on the same request is rejected
explicitly rather than producing a duplicate Beancount entry."""
_, wallet = configured_user
submitted = await submit_manual_payment_request(
client,
wallet_inkey=wallet.inkey,
amount_sats=7_500,
description=f"approve once {uuid4().hex[:6]}",
)
await approve_manual_payment_request(
client, super_user_headers=super_user_headers, request_id=submitted["id"],
)
r = await client.post(
f"/libra/api/v1/manual-payment-requests/{submitted['id']}/approve",
headers=super_user_headers,
)
assert r.status_code == 400, f"expected 400, got {r.status_code}: {r.text}"
assert "approve" in r.text.lower()

294
tests/test_reconciliation_api.py
View file

@ -1,294 +0,0 @@
"""Balance assertion CRUD + reconciliation summary endpoints.
Endpoints:
- `POST /libra/api/v1/assertions` — create + check
- `GET /libra/api/v1/assertions` — list with filters
- `GET /libra/api/v1/assertions/{id}` — fetch one
- `POST /libra/api/v1/assertions/{id}/check` — re-check
- `DELETE /libra/api/v1/assertions/{id}` — remove
All `require_super_user` (libra-level, wallet admin-key).
The create endpoint is hybrid: it posts a Beancount `balance` directive via
Fava (source of truth), persists the assertion metadata in libra's DB, and
re-checks immediately. On mismatch it returns 409 with the diff payload.
"""
from uuid import uuid4
import pytest
# Tests that try to actually create + check an assertion all hit issue #39:
# `format_balance` returns a Beancount source string but `fava.add_entry`
# expects a dict, so Fava 500s on every assertion-create call. The contract
# violation is on libra's side; mark these strict-xfail so they go green
# automatically once #39 lands and the format_balance return shape is fixed.
ASSERTION_CREATE_BROKEN = pytest.mark.xfail(
reason="libra/issues/39 — POST /assertions submits a Beancount source string "
"to Fava's JSON API and 500s. Drop this marker when the format_balance "
"return type is changed to a dict.",
strict=True,
)
# ---------------------------------------------------------------------------
# helpers (local — assertion endpoints don't have wrapper helpers yet)
# ---------------------------------------------------------------------------
async def _create_assertion(
client, *, super_user_headers, account_id, expected_sats,
tolerance_sats=0, fiat_currency=None, expected_fiat=None,
):
body = {
"account_id": account_id,
"expected_balance_sats": expected_sats,
"tolerance_sats": tolerance_sats,
}
if fiat_currency:
body["fiat_currency"] = fiat_currency
body["expected_balance_fiat"] = str(expected_fiat) if expected_fiat is not None else "0"
return await client.post(
"/libra/api/v1/assertions", headers=super_user_headers, json=body,
)
# ---------------------------------------------------------------------------
# tests
# ---------------------------------------------------------------------------
@ASSERTION_CREATE_BROKEN
@pytest.mark.anyio
async def test_assertion_against_empty_account_passes(
client, super_user_headers, standard_accounts,
):
"""An asset account with no postings has a 0 balance — asserting 0
should pass and the resulting assertion has status='passed'."""
r = await _create_assertion(
client,
super_user_headers=super_user_headers,
account_id=standard_accounts["assets_cash"]["id"],
expected_sats=0,
)
assert r.status_code == 200, f"expected 200, got {r.status_code}: {r.text}"
body = r.json()
assert body.get("status") == "passed", (
f"expected status='passed' for 0=0, got {body.get('status')} body={body}"
)
assert body.get("difference_sats", 0) == 0
@ASSERTION_CREATE_BROKEN
@pytest.mark.anyio
async def test_assertion_with_wrong_balance_returns_409(
client, super_user_headers, standard_accounts,
):
"""When the actual balance doesn't match expected, the create endpoint
returns 409 Conflict with the diff payload — Beancount validates it
server-side after the directive lands."""
r = await _create_assertion(
client,
super_user_headers=super_user_headers,
account_id=standard_accounts["assets_cash"]["id"],
expected_sats=999_999, # wildly wrong for empty account
)
assert r.status_code == 409, f"expected 409, got {r.status_code}: {r.text}"
# 409 body should expose the diff so a UI can render the gap.
detail = r.json().get("detail")
assert isinstance(detail, dict), f"expected structured detail, got {detail!r}"
assert detail.get("expected_sats") == 999_999
assert detail.get("actual_sats") == 0
assert detail.get("difference_sats") == 999_999 or detail.get("difference_sats") == -999_999
@ASSERTION_CREATE_BROKEN
@pytest.mark.anyio
async def test_assertion_with_tolerance_accepts_small_diff(
client, super_user_headers, standard_accounts,
):
"""A tolerance of N sats lets actual-vs-expected diverge by ≤N."""
r = await _create_assertion(
client,
super_user_headers=super_user_headers,
account_id=standard_accounts["assets_cash"]["id"],
expected_sats=50,
tolerance_sats=100, # actual=0, expected=50, diff=50, tolerance=100 → passes
)
assert r.status_code == 200, f"expected 200 within tolerance, got {r.status_code}: {r.text}"
assert r.json().get("status") == "passed"
@ASSERTION_CREATE_BROKEN
@pytest.mark.anyio
async def test_list_assertions_returns_created(
client, super_user_headers, standard_accounts,
):
"""Newly created assertions show up in the list filtered by account."""
account_id = standard_accounts["assets_cash"]["id"]
create = await _create_assertion(
client,
super_user_headers=super_user_headers,
account_id=account_id,
expected_sats=0,
)
assert create.status_code == 200
assertion_id = create.json()["id"]
r = await client.get(
f"/libra/api/v1/assertions?account_id={account_id}",
headers=super_user_headers,
)
assert r.status_code == 200, f"list assertions: {r.status_code} {r.text}"
ids = [a.get("id") for a in r.json()]
assert assertion_id in ids, f"created assertion {assertion_id} missing from list {ids}"
@ASSERTION_CREATE_BROKEN
@pytest.mark.anyio
async def test_get_assertion_by_id(
client, super_user_headers, standard_accounts,
):
create = await _create_assertion(
client,
super_user_headers=super_user_headers,
account_id=standard_accounts["assets_cash"]["id"],
expected_sats=0,
)
assert create.status_code == 200
assertion_id = create.json()["id"]
r = await client.get(
f"/libra/api/v1/assertions/{assertion_id}",
headers=super_user_headers,
)
assert r.status_code == 200, f"get assertion: {r.status_code} {r.text}"
assert r.json().get("id") == assertion_id
@ASSERTION_CREATE_BROKEN
@pytest.mark.anyio
async def test_recheck_assertion_via_check_endpoint(
client, super_user_headers, standard_accounts,
):
"""`POST /assertions/{id}/check` re-evaluates and returns the updated
assertion record. Idempotent against a stable ledger state."""
create = await _create_assertion(
client,
super_user_headers=super_user_headers,
account_id=standard_accounts["assets_cash"]["id"],
expected_sats=0,
)
assertion_id = create.json()["id"]
r = await client.post(
f"/libra/api/v1/assertions/{assertion_id}/check",
headers=super_user_headers,
)
assert r.status_code == 200, f"recheck: {r.status_code} {r.text}"
assert r.json().get("status") == "passed"
@ASSERTION_CREATE_BROKEN
@pytest.mark.anyio
async def test_delete_assertion_removes_it(
client, super_user_headers, standard_accounts,
):
create = await _create_assertion(
client,
super_user_headers=super_user_headers,
account_id=standard_accounts["assets_cash"]["id"],
expected_sats=0,
)
assertion_id = create.json()["id"]
r = await client.delete(
f"/libra/api/v1/assertions/{assertion_id}",
headers=super_user_headers,
)
assert r.status_code in (200, 204), f"delete: {r.status_code} {r.text}"
# Subsequent GET should 404.
r = await client.get(
f"/libra/api/v1/assertions/{assertion_id}",
headers=super_user_headers,
)
assert r.status_code == 404, f"expected 404 after delete, got {r.status_code}"
@pytest.mark.anyio
async def test_assertion_unknown_account_returns_404(
client, super_user_headers,
):
"""Account-not-found check happens before any Beancount write."""
r = await _create_assertion(
client,
super_user_headers=super_user_headers,
account_id=f"nonexistent-{uuid4().hex[:6]}",
expected_sats=0,
)
assert r.status_code == 404, f"expected 404, got {r.status_code}: {r.text}"
@pytest.mark.anyio
async def test_non_super_user_cannot_create_assertion(
client, configured_user, standard_accounts,
):
"""Wallet admin-key of a regular user fails the super-user identity
check — 403."""
_, wallet = configured_user
r = await client.post(
"/libra/api/v1/assertions",
headers={"X-Api-Key": wallet.adminkey, "Content-type": "application/json"},
json={
"account_id": standard_accounts["assets_cash"]["id"],
"expected_balance_sats": 0,
},
)
assert r.status_code == 403, f"expected 403, got {r.status_code}: {r.text}"
assert "super" in r.text.lower()
@pytest.mark.anyio
async def test_list_assertions_invalid_status_returns_400(
client, super_user_headers,
):
"""Status filter is validated against the AssertionStatus enum."""
r = await client.get(
"/libra/api/v1/assertions?status=not_a_status",
headers=super_user_headers,
)
assert r.status_code == 400, f"expected 400, got {r.status_code}: {r.text}"
assert "status" in r.text.lower()
@pytest.mark.anyio
async def test_reconciliation_summary_endpoint(client, super_user_headers):
"""`GET /reconciliation/summary` responds 200 and returns a structured
payload even when no assertions exist. Smoke-shape only — exact counts
depend on ledger history.
Doesn't pre-create an assertion (#39 blocks that path); the summary
endpoint should still serve a default empty shape.
"""
r = await client.get(
"/libra/api/v1/reconciliation/summary",
headers=super_user_headers,
)
assert r.status_code == 200, f"reconciliation summary: {r.status_code} {r.text}"
payload = r.json()
assert isinstance(payload, dict), f"expected dict, got {type(payload)}"
@pytest.mark.anyio
async def test_daily_reconciliation_task_runs(
client, super_user_headers,
):
"""The daily-reconciliation task endpoint returns 200 even when no
assertions exist — it's the entry point that ops cron hits."""
r = await client.post(
"/libra/api/v1/tasks/daily-reconciliation",
headers=super_user_headers,
)
assert r.status_code == 200, f"daily-reconciliation: {r.status_code} {r.text}"

202
tests/test_settings_auth_api.py
View file

@ -1,202 +0,0 @@
"""Settings and per-user wallet endpoints, plus the auth gates around them.
Endpoints and their auth profiles:
- `GET /libra/api/v1/settings` — any authenticated user.
- `PUT /libra/api/v1/settings` — `check_super_user` (Bearer, super-user only).
- `GET /libra/api/v1/user/wallet` — `check_user_exists` (any authed user).
- `PUT /libra/api/v1/user/wallet` — `check_user_exists`.
- `GET /libra/api/v1/user-wallet/{user_id}` — `require_super_user` (libra
super-user via wallet admin-key auth).
Two distinct super-user auth flows live here side by side:
- LNbits-level `check_super_user` → Bearer token from username/password login.
- Libra-level `require_super_user` → wallet admin-key of the super-user-owned
wallet.
Tests use the `super_user_bearer_headers` fixture for the first, the
`super_user_headers` fixture for the second, and `?usr=<user_id>` for
non-admin authed calls.
"""
from uuid import uuid4
import pytest
@pytest.mark.anyio
async def test_super_user_can_get_and_update_settings(
client, super_user_bearer_headers, libra_wallet, fava_process,
):
"""Super user round-trips through `GET /settings` → mutate → `PUT /settings`.
Verifies the Bearer-auth happy path and confirms `update_settings`
persists what we sent (modulo defaults libra fills in).
"""
r = await client.get(
"/libra/api/v1/settings", headers=super_user_bearer_headers,
)
assert r.status_code == 200, f"GET /settings: {r.status_code} {r.text}"
original = r.json()
assert original.get("libra_wallet_id") == libra_wallet.id, (
f"libra_wallet fixture should have configured wallet_id, got {original}"
)
new_timeout = 7.5
r = await client.put(
"/libra/api/v1/settings",
headers=super_user_bearer_headers,
json={
"libra_wallet_id": libra_wallet.id,
"fava_url": fava_process,
"fava_ledger_slug": "libra-test",
"fava_timeout": new_timeout,
},
)
assert r.status_code == 200, f"PUT /settings: {r.status_code} {r.text}"
assert float(r.json().get("fava_timeout", 0)) == pytest.approx(new_timeout)
# Reset to keep other tests' baseline intact.
await client.put(
"/libra/api/v1/settings",
headers=super_user_bearer_headers,
json={
"libra_wallet_id": libra_wallet.id,
"fava_url": fava_process,
"fava_ledger_slug": "libra-test",
"fava_timeout": original.get("fava_timeout", 5.0),
},
)
@pytest.mark.anyio
async def test_put_settings_without_libra_wallet_id_returns_400(
client, super_user_bearer_headers,
):
"""The settings endpoint explicitly rejects updates with no wallet id.
This is the validation libra applies before any persistence so we don't
silently accept a settings row that breaks all entry endpoints.
"""
r = await client.put(
"/libra/api/v1/settings",
headers=super_user_bearer_headers,
json={"fava_url": "http://example.test"},
)
assert r.status_code == 400, f"expected 400, got {r.status_code}: {r.text}"
assert "wallet" in r.text.lower()
@pytest.mark.anyio
async def test_put_settings_without_auth_returns_401(client, libra_wallet):
"""No auth at all → LNbits's `check_admin` rejects with 401."""
r = await client.put(
"/libra/api/v1/settings",
json={"libra_wallet_id": libra_wallet.id},
)
assert r.status_code == 401, f"expected 401, got {r.status_code}: {r.text}"
@pytest.mark.anyio
async def test_regular_user_cannot_put_settings(
client, configured_user, libra_wallet,
):
"""A non-super user (regardless of auth method they try) cannot update
libra settings. Using `?usr=<id>` to mimic user-id login."""
user, _ = configured_user
r = await client.put(
f"/libra/api/v1/settings?usr={user.id}",
json={"libra_wallet_id": libra_wallet.id},
)
# `_check_account_exists` forbids user-id login for admin accounts and
# rejects regular users from `check_admin` paths — either 401 or 403
# is a valid no-access response here.
assert r.status_code in (401, 403), (
f"expected 401/403, got {r.status_code}: {r.text}"
)
@pytest.mark.anyio
async def test_regular_user_can_get_and_update_own_user_wallet(
client, libra_user, libra_wallet, # noqa: ARG001 (libra_wallet ensures session setup)
):
"""A regular user (no admin perm) can read and update their own
`user_wallet_id` via `?usr=<id>`."""
user, wallet = libra_user
r = await client.get(f"/libra/api/v1/user/wallet?usr={user.id}")
assert r.status_code == 200, f"GET /user/wallet: {r.status_code} {r.text}"
r = await client.put(
f"/libra/api/v1/user/wallet?usr={user.id}",
json={"user_wallet_id": wallet.id},
)
assert r.status_code == 200, f"PUT /user/wallet: {r.status_code} {r.text}"
r = await client.get(f"/libra/api/v1/user/wallet?usr={user.id}")
assert r.json().get("user_wallet_id") == wallet.id, (
f"GET after PUT should echo wallet id, got {r.json()}"
)
@pytest.mark.anyio
async def test_super_user_can_get_any_user_wallet(
client, super_user_headers, configured_user,
):
"""The `/user-wallet/{user_id}` endpoint (libra `require_super_user`,
wallet-admin-key auth) returns wallet info for any user."""
user, wallet = configured_user
r = await client.get(
f"/libra/api/v1/user-wallet/{user.id}", headers=super_user_headers,
)
assert r.status_code == 200, f"GET /user-wallet/{user.id}: {r.status_code} {r.text}"
payload = r.json()
assert payload.get("user_id") == user.id
assert payload.get("user_wallet_id") == wallet.id, (
f"expected user_wallet_id={wallet.id}, got {payload}"
)
@pytest.mark.anyio
async def test_regular_user_cannot_use_super_only_user_wallet_endpoint(
client, configured_user, configured_user_b,
):
"""User B can't see user A's wallet info via the super-only admin
endpoint, even with B's own wallet admin-key."""
user_a, _ = configured_user
_, wallet_b = configured_user_b
r = await client.get(
f"/libra/api/v1/user-wallet/{user_a.id}",
headers={"X-Api-Key": wallet_b.adminkey, "Content-type": "application/json"},
)
assert r.status_code == 403, f"expected 403, got {r.status_code}: {r.text}"
assert "super" in r.text.lower()
@pytest.mark.anyio
async def test_unknown_currency_in_settings_does_not_corrupt(
client, super_user_bearer_headers, libra_wallet, fava_process,
):
"""Passing an unexpected field in the settings body shouldn't bring the
endpoint down — pydantic should ignore extras and persist the rest.
A canary for "what if the UI sends a slightly-stale settings shape?"
"""
r = await client.put(
"/libra/api/v1/settings",
headers=super_user_bearer_headers,
json={
"libra_wallet_id": libra_wallet.id,
"fava_url": fava_process,
"fava_ledger_slug": "libra-test",
"some_unexpected_field_": str(uuid4()),
},
)
# Either 200 (extras dropped) or 422 (strict validation) — both are
# acceptable defensive behaviours; just don't 500.
assert r.status_code in (200, 422), (
f"unexpected field should be ignored or rejected cleanly, "
f"got {r.status_code}: {r.text}"
)

342
tests/test_settlement_api.py
View file

@ -1,342 +0,0 @@
"""Settlement netting + credit overflow — libra-#33 + libra-#41.
`POST /libra/api/v1/receivables/settle` with `settled_entry_links=None`
(the default) auto-detects open entries in both directions, builds a
3-leg settlement transaction that zeros out both per-user accounts when
the user has open balances on both sides (libra-#33's nancy scenario),
and routes any excess cash to `Liabilities:Credit:User-X` (libra-#41).
Underpay without explicit entry-picks returns 400 with diff details so
the operator can either pay the exact net or specify `settled_entry_links`.
"""
import importlib
from uuid import uuid4
import pytest
from .helpers import (
approve_entry,
get_balance,
list_user_entries,
post_expense,
post_receivable,
settle_receivable,
)
def _libra_module(submodule: str):
for prefix in ("lnbits.extensions.libra", "libra"):
try:
return importlib.import_module(f"{prefix}.{submodule}")
except ModuleNotFoundError:
continue
raise ModuleNotFoundError(f"libra.{submodule}")
async def _approve_and_refresh(client, wallet, super_user_headers, entry_id):
"""Approve a pending entry and force a Fava reload (libra-#37 workaround)."""
await approve_entry(client, super_user_headers=super_user_headers, entry_id=entry_id)
await list_user_entries(client, wallet_inkey=wallet.inkey)
# ---------------------------------------------------------------------------
# Nancy's #33 scenario and variants
# ---------------------------------------------------------------------------
@pytest.mark.anyio
async def test_exact_net_settlement_zeroes_both_per_user_accounts(
client, super_user_headers, configured_user, standard_accounts,
):
"""Nancy: receivable 100 EUR + payable 50 EUR + 50 EUR cash → 3-leg
settlement that zeros both Receivable and Payable for this user.
Acceptance criteria from libra-#33:
- Settlement links every source entry it reconciles.
- Per-user balances drop to 0 (not just net to 0 leaving each side open).
"""
user, wallet = configured_user
tag = uuid4().hex[:6]
# Admin records the receivable (cleared on creation).
await post_receivable(
client,
super_user_headers=super_user_headers,
user_id=user.id,
amount="100.00", currency="EUR",
description=f"Rent share {tag}",
revenue_account=standard_accounts["revenue_rent"]["name"],
)
# User submits an expense (pending until admin approves).
exp = await post_expense(
client,
wallet_inkey=wallet.inkey,
user_wallet_id=wallet.id,
amount="50.00", currency="EUR",
description=f"Drill purchase {tag}",
expense_account=standard_accounts["expense_food"]["name"],
)
await _approve_and_refresh(client, wallet, super_user_headers, exp["id"])
# Sanity check: user owes 50 EUR net (100 receivable - 50 payable).
balance_before = await get_balance(client, wallet_inkey=wallet.inkey)
eur_before = balance_before.get("fiat_balances", {}).get("EUR")
assert float(eur_before) == pytest.approx(50.0), (
f"expected +50 EUR net (user owes libra), got {eur_before}"
)
# Settle the net cash: 50 EUR.
await settle_receivable(
client,
super_user_headers=super_user_headers,
user_id=user.id,
amount="50.00", currency="EUR",
description=f"Cash settlement {tag}",
)
await list_user_entries(client, wallet_inkey=wallet.inkey)
# After settlement: net balance is 0.
balance_after = await get_balance(client, wallet_inkey=wallet.inkey)
eur_after = balance_after.get("fiat_balances", {}).get("EUR", 0)
assert float(eur_after or 0) == pytest.approx(0.0), (
f"expected 0 EUR after exact net settlement, got {eur_after}"
)
# Per-account breakdown: every user-side account is at 0.
# (The acceptance criterion is that NEITHER Receivable nor Payable
# carries an open balance — not just that they net to 0.)
breakdown = balance_after.get("account_balances", [])
for row in breakdown:
if user.id[:8] in (row.get("account") or ""):
assert float(row.get("eur", 0) or 0) == pytest.approx(0.0), (
f"per-user account {row['account']} still has "
f"{row.get('eur')} EUR open after complete settlement; "
f"libra-#33 acceptance criterion violated"
)
# The settlement entry's links must cover both source entries.
# Both rcv-* and exp-* links should appear via Fava query.
fava_client_mod = _libra_module("fava_client")
fava = fava_client_mod.get_fava_client()
unsettled_receivables = await fava.get_unsettled_entries_bql(user.id, "receivable")
unsettled_payables = await fava.get_unsettled_entries_bql(user.id, "expense")
assert not unsettled_receivables, (
f"receivable left as unsettled after complete settlement: "
f"{unsettled_receivables}"
)
assert not unsettled_payables, (
f"payable left as unsettled after complete settlement: "
f"{unsettled_payables}"
)
@pytest.mark.anyio
async def test_overpay_routes_excess_to_credit(
client, super_user_headers, configured_user, standard_accounts,
):
"""Receivable 100 + payable 50 + cash 70 EUR → settles both per-user
accounts to 0, and the 20 EUR excess lands on Liabilities:Credit:User-X
(libra now owes the user 20 going forward).
Headline libra-#41 case: cash > net obligation absorbed into credit.
"""
user, wallet = configured_user
tag = uuid4().hex[:6]
await post_receivable(
client,
super_user_headers=super_user_headers,
user_id=user.id,
amount="100.00", currency="EUR",
description=f"Receivable {tag}",
revenue_account=standard_accounts["revenue_rent"]["name"],
)
exp = await post_expense(
client,
wallet_inkey=wallet.inkey,
user_wallet_id=wallet.id,
amount="50.00", currency="EUR",
description=f"Payable {tag}",
expense_account=standard_accounts["expense_food"]["name"],
)
await _approve_and_refresh(client, wallet, super_user_headers, exp["id"])
# User pays 70 EUR — 20 EUR over the 50 EUR net obligation.
await settle_receivable(
client,
super_user_headers=super_user_headers,
user_id=user.id,
amount="70.00", currency="EUR",
description=f"Overpay settlement {tag}",
)
await list_user_entries(client, wallet_inkey=wallet.inkey)
# Net balance should be -20 EUR (libra owes user 20 via credit).
balance = await get_balance(client, wallet_inkey=wallet.inkey)
eur = balance.get("fiat_balances", {}).get("EUR")
assert float(eur) == pytest.approx(-20.0), (
f"expected -20 EUR (libra owes user via credit), got {eur} from {balance}"
)
# Credit account should appear in the breakdown with -20 EUR.
breakdown = balance.get("account_balances", [])
credit_row = next(
(r for r in breakdown if "Credit" in (r.get("account") or "")), None,
)
assert credit_row is not None, (
f"Credit account missing from breakdown: {breakdown}"
)
assert float(credit_row.get("eur", 0)) == pytest.approx(-20.0), (
f"expected -20 EUR on Credit:User-X, got {credit_row.get('eur')}"
)
@pytest.mark.anyio
async def test_pure_receivable_overpay_creates_credit(
client, super_user_headers, configured_user, standard_accounts,
):
"""No payable side — receivable 50 + cash 70 → receivable cleared,
20 EUR moves to credit. 2-leg + credit overflow leg."""
user, wallet = configured_user
tag = uuid4().hex[:6]
await post_receivable(
client,
super_user_headers=super_user_headers,
user_id=user.id,
amount="50.00", currency="EUR",
description=f"Pure receivable {tag}",
revenue_account=standard_accounts["revenue_rent"]["name"],
)
await list_user_entries(client, wallet_inkey=wallet.inkey)
await settle_receivable(
client,
super_user_headers=super_user_headers,
user_id=user.id,
amount="70.00", currency="EUR",
description=f"Pure overpay {tag}",
)
await list_user_entries(client, wallet_inkey=wallet.inkey)
balance = await get_balance(client, wallet_inkey=wallet.inkey)
eur = balance.get("fiat_balances", {}).get("EUR")
# Receivable cleared (0) - credit (-20) = -20 net
assert float(eur) == pytest.approx(-20.0), (
f"expected -20 EUR after pure overpay, got {eur}"
)
# ---------------------------------------------------------------------------
# Validation: underpay without explicit links → 400 with diff
# ---------------------------------------------------------------------------
@pytest.mark.anyio
async def test_underpay_without_explicit_links_returns_400(
client, super_user_headers, configured_user, standard_accounts,
):
"""Cash < net obligation and no `settled_entry_links` → 400 with the
diff payload so operator can fix the amount or specify entries.
Without #41's credit overflow + #33's auto-detect, this was the
silent-drift case that motivated both issues. Now: explicit, recoverable.
"""
user, wallet = configured_user
await post_receivable(
client,
super_user_headers=super_user_headers,
user_id=user.id,
amount="100.00", currency="EUR",
description="Receivable to underpay against",
revenue_account=standard_accounts["revenue_rent"]["name"],
)
await list_user_entries(client, wallet_inkey=wallet.inkey)
r = await client.post(
"/libra/api/v1/receivables/settle",
headers=super_user_headers,
json={
"user_id": user.id,
"amount": "30.00",
"currency": "EUR",
"payment_method": "cash",
"description": "Underpay attempt",
},
)
assert r.status_code == 400, f"expected 400, got {r.status_code}: {r.text}"
payload = r.json().get("detail")
assert isinstance(payload, dict), f"expected structured detail, got {payload!r}"
assert payload.get("cash_paid") == 30.0
assert payload.get("net_obligation") == 100.0
assert payload.get("receivable_total") == 100.0
assert payload.get("payable_total") == 0.0
@pytest.mark.anyio
async def test_no_open_receivable_returns_400(
client, super_user_headers, configured_user,
):
"""User has no open receivables → endpoint can't settle. 400 with a
hint pointing at `/payables/pay` for the inverse direction."""
user, _ = configured_user
r = await client.post(
"/libra/api/v1/receivables/settle",
headers=super_user_headers,
json={
"user_id": user.id,
"amount": "50.00",
"currency": "EUR",
"payment_method": "cash",
"description": "Random deposit attempt",
},
)
assert r.status_code == 400, f"expected 400, got {r.status_code}: {r.text}"
assert "no open receivables" in r.text.lower() or "payables/pay" in r.text
# ---------------------------------------------------------------------------
# Legacy explicit-links path: preserved for partial-settle-of-specific-entries
# ---------------------------------------------------------------------------
@pytest.mark.anyio
async def test_explicit_settled_entry_links_uses_legacy_2_leg_path(
client, super_user_headers, configured_user, standard_accounts,
):
"""When `settled_entry_links` is provided, backend trusts the caller's
list and writes the legacy 2-leg shape. No auto-netting, no credit
overflow validation. Required for callers that want to settle a
specific subset of entries.
Requires `amount_sats` per the legacy path's existing contract.
"""
user, wallet = configured_user
await post_receivable(
client,
super_user_headers=super_user_headers,
user_id=user.id,
amount="50.00", currency="EUR",
description="Receivable for explicit-link test",
revenue_account=standard_accounts["revenue_rent"]["name"],
)
await list_user_entries(client, wallet_inkey=wallet.inkey)
# Caller passes explicit (but possibly empty) link list → legacy path.
r = await client.post(
"/libra/api/v1/receivables/settle",
headers=super_user_headers,
json={
"user_id": user.id,
"amount": "50.00",
"currency": "EUR",
"amount_sats": 55_000,
"payment_method": "cash",
"description": "Explicit-link settle",
"settled_entry_links": [], # opts out of auto-detect
},
)
assert r.status_code == 200, f"legacy explicit-link path: {r.status_code} {r.text}"

66
tests/test_smoke.py
View file

@ -1,66 +0,0 @@
"""Smoke test: validates the test harness end-to-end.
If this passes, the rest of the test files can be trusted to actually exercise
real code paths (Fava up, app up, Libra activated, FavaClient pointed at the
test instance, BQL round-trips working, libra wallet configured, user wallet
configured, account exists, permission granted).
If this fails, no point running anything else — fix the harness first.
"""
import pytest
from .helpers import approve_entry, get_balance, post_expense
@pytest.mark.anyio
async def test_smoke_submit_approve_and_see_balance(
client, super_user_headers, configured_user, standard_accounts,
):
"""Full stack round-trip: user submits an expense, admin approves it,
balance reflects it.
Exercises: libra wallet config (session fixture), user wallet config
(configured_user fixture), permission grant (configured_user fixture),
Beancount entry construction, Fava add_entries HTTP call, pending→cleared
flag transition via the source-slice mutation path, BQL balance query
(which filters by flag = '*' so the approve step is load-bearing).
"""
_, wallet = configured_user
# User pays 50 EUR for groceries — entry posted with flag `!` (pending).
entry = await post_expense(
client,
wallet_inkey=wallet.inkey,
user_wallet_id=wallet.id,
amount="50.00",
currency="EUR",
description="Smoke test expense",
expense_account=standard_accounts["expense_food"]["name"],
)
entry_id = entry.get("id")
assert entry_id, f"expense response missing id: {entry}"
# Pending entries are excluded from the cleared-only balance query —
# confirm balance is still zero at this point.
pending_balance = await get_balance(client, wallet_inkey=wallet.inkey)
pending_eur = pending_balance.get("fiat_balances", {}).get("EUR")
assert pending_eur in (None, 0, "0", "0.00"), (
f"pending expense should not affect cleared balance, got {pending_eur}"
)
# Admin approves the pending entry, flipping its flag from `!` to `*`.
await approve_entry(
client, super_user_headers=super_user_headers, entry_id=entry_id,
)
# Balance now reflects the 50 EUR Libra owes the user.
# Sign convention (per get_user_balance_bql docstring): the API returns
# the balance from libra's perspective — negative on Liabilities:Payable
# means libra owes the user. So a 50 EUR expense surfaces as -50 EUR.
balance = await get_balance(client, wallet_inkey=wallet.inkey)
fiat = balance.get("fiat_balances", {})
eur = fiat.get("EUR")
assert eur is not None, f"expected EUR in fiat_balances after approve, got {balance}"
assert float(eur) == pytest.approx(-50.0), (
f"expected EUR balance of -50.00 (libra-owes-user) after approve, got {eur}"
)

416
tests/test_unit.py
View file

@ -1,416 +0,0 @@
"""Pure-function unit tests — no harness, no Fava, no LNbits app.
Covers `libra.beancount_format`, `libra.account_utils`, `libra.core.validation`.
These modules have no external dependencies (stdlib + pydantic for models), so
they run fast and don't need fixtures.
The libra package is importable under either `lnbits.extensions.libra.*`
(default lnbits layout) or `libra.*` (LNBITS_EXTENSIONS_PATH override). The
`_module` helper tries both, mirroring the runtime-path discipline already
established in `conftest.py`.
"""
import importlib
from datetime import date
from decimal import Decimal
import pytest
def _module(name: str):
"""Import a libra submodule under whichever path the active LNbits layout
uses (default `lnbits.extensions.libra` or bare `libra`)."""
for prefix in ("lnbits.extensions.libra", "libra"):
try:
return importlib.import_module(f"{prefix}.{name}")
except ModuleNotFoundError:
continue
raise ModuleNotFoundError(f"libra.{name}: tried both import paths")
bf = _module("beancount_format")
au = _module("account_utils")
val = _module("core.validation")
mdl = _module("models")
AccountType = mdl.AccountType
# ---------------------------------------------------------------------------
# beancount_format.sanitize_link
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
("raw", "expected"),
[
("libra-abc123", "libra-abc123"),
("Invoice #123", "Invoice-123"),
("Test (pending)", "Test-pending"),
("a/b.c-d_e", "a/b.c-d_e"), # all permitted chars survive
("multiple spaces", "multiple-spaces"), # collapsed
("---leading-trailing---", "leading-trailing"),
("ascii_only", "ascii_only"),
],
)
def test_sanitize_link_strips_unsafe_chars(raw, expected):
assert bf.sanitize_link(raw) == expected
def test_sanitize_link_empty_string_stays_empty():
assert bf.sanitize_link("") == ""
def test_sanitize_link_unicode_replaced_with_hyphens():
# Non-ascii chars all collapse to single hyphens, stripped from edges.
result = bf.sanitize_link("café résumé")
assert all(ch in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-_/."
for ch in result), f"unsanitized chars in {result!r}"
assert not result.startswith("-")
assert not result.endswith("-")
# ---------------------------------------------------------------------------
# beancount_format.format_transaction
# ---------------------------------------------------------------------------
def test_format_transaction_minimum_shape():
entry = bf.format_transaction(
date_val=date(2026, 6, 6),
flag="*",
narration="hello",
postings=[{"account": "Assets:Cash", "amount": "10 EUR"}],
)
# Fava's required fields.
assert entry["t"] == "Transaction"
assert entry["date"] == "2026-06-06"
assert entry["flag"] == "*"
assert entry["narration"] == "hello"
assert entry["payee"] == "" # empty string, not None
assert entry["tags"] == []
assert entry["links"] == []
assert entry["meta"] == {}
assert entry["postings"] == [{"account": "Assets:Cash", "amount": "10 EUR"}]
def test_format_transaction_optional_fields_are_passed_through():
entry = bf.format_transaction(
date_val=date(2026, 6, 6),
flag="!",
narration="pending lunch",
postings=[{"account": "Expenses:Food", "amount": "8 EUR"}],
payee="Bistro Local",
tags=["expense-entry"],
links=["libra-abc123"],
meta={"user-id": "abc12345"},
)
assert entry["flag"] == "!"
assert entry["payee"] == "Bistro Local"
assert entry["tags"] == ["expense-entry"]
assert entry["links"] == ["libra-abc123"]
assert entry["meta"] == {"user-id": "abc12345"}
def test_format_transaction_does_not_share_mutable_defaults():
"""Regression guard: passing `tags=None` shouldn't return the same list
every call (the classic Python mutable-default-argument trap)."""
a = bf.format_transaction(date(2026, 1, 1), "*", "a", [{"account": "X", "amount": "1 EUR"}])
b = bf.format_transaction(date(2026, 1, 2), "*", "b", [{"account": "Y", "amount": "1 EUR"}])
a["tags"].append("touched-a")
assert b["tags"] == [], "tags from one entry leaked into another"
# ---------------------------------------------------------------------------
# beancount_format.generate_entry_id
# ---------------------------------------------------------------------------
def test_generate_entry_id_shape():
eid = bf.generate_entry_id()
assert len(eid) == 16
assert all(c in "0123456789abcdef" for c in eid), f"non-hex in {eid!r}"
def test_generate_entry_ids_are_unique():
ids = {bf.generate_entry_id() for _ in range(100)}
assert len(ids) == 100 # 16 hex chars = 64 bits; collisions in 100 are negligible
# ---------------------------------------------------------------------------
# account_utils.format_hierarchical_account_name
# ---------------------------------------------------------------------------
def test_format_hierarchical_simple_asset():
assert au.format_hierarchical_account_name(AccountType.ASSET, "Cash") == "Assets:Cash"
def test_format_hierarchical_user_specific_uses_8_char_prefix():
full_user_id = "af983632aabbccddeeff00112233445566"
name = au.format_hierarchical_account_name(
AccountType.ASSET, "Accounts Receivable", user_id=full_user_id,
)
assert name == "Assets:Receivable:User-af983632" # 8-char prefix, "Accounts " stripped
def test_format_hierarchical_ampersand_expands_to_colon():
"""`Food & Supplies` is a legacy display form; it becomes a hierarchy."""
name = au.format_hierarchical_account_name(AccountType.EXPENSE, "Food & Supplies")
assert name == "Expenses:Food:Supplies"
def test_format_hierarchical_revenue_uses_income_root():
"""Beancount uses `Income`, not `Revenue` — the mapping is in
`ACCOUNT_TYPE_ROOTS`."""
name = au.format_hierarchical_account_name(AccountType.REVENUE, "Accommodation")
assert name == "Income:Accommodation"
# ---------------------------------------------------------------------------
# account_utils.parse_legacy_account_name
# ---------------------------------------------------------------------------
def test_parse_legacy_with_user_suffix():
assert au.parse_legacy_account_name("Accounts Receivable - af983632") == (
"Accounts Receivable", "af983632",
)
def test_parse_legacy_without_user_suffix():
assert au.parse_legacy_account_name("Cash") == ("Cash", None)
# ---------------------------------------------------------------------------
# account_utils.format_account_display_name
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
("hierarchical", "expected"),
[
("Assets:Receivable:User-af983632", "Accounts Receivable - af983632"),
("Liabilities:Payable:User-cafebabe", "Accounts Payable - cafebabe"),
("Expenses:Food:Supplies", "Food & Supplies"),
("Assets:Cash", "Cash"),
("Assets", "Assets"), # too short — passes through
],
)
def test_format_account_display_name(hierarchical, expected):
assert au.format_account_display_name(hierarchical) == expected
# ---------------------------------------------------------------------------
# account_utils.get_account_type_from_hierarchical
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
("name", "expected_type"),
[
("Assets:Cash", AccountType.ASSET),
("Liabilities:Payable:User-x", AccountType.LIABILITY),
("Equity:User-x", AccountType.EQUITY),
("Income:Accommodation", AccountType.REVENUE),
("Expenses:Food", AccountType.EXPENSE),
],
)
def test_get_account_type_from_hierarchical(name, expected_type):
assert au.get_account_type_from_hierarchical(name) == expected_type
def test_get_account_type_unknown_root_returns_none():
assert au.get_account_type_from_hierarchical("Other:Random") is None
# ---------------------------------------------------------------------------
# account_utils.migrate_account_name — round-trip legacy → hierarchical
# ---------------------------------------------------------------------------
def test_migrate_account_name_receivable():
out = au.migrate_account_name("Accounts Receivable - af983632", AccountType.ASSET)
assert out == "Assets:Receivable:User-af983632"
def test_migrate_account_name_expense_with_ampersand():
assert au.migrate_account_name("Food & Supplies", AccountType.EXPENSE) == (
"Expenses:Food:Supplies"
)
# ---------------------------------------------------------------------------
# core.validation — validate_journal_entry
# ---------------------------------------------------------------------------
def test_validate_journal_entry_balanced_passes():
val.validate_journal_entry(
{"id": "x"},
[
{"account_id": "a", "amount": 100},
{"account_id": "b", "amount": -100},
],
)
def test_validate_journal_entry_unbalanced_raises():
with pytest.raises(val.ValidationError) as exc:
val.validate_journal_entry(
{"id": "x"},
[
{"account_id": "a", "amount": 100},
{"account_id": "b", "amount": -50},
],
)
assert "not balanced" in str(exc.value)
def test_validate_journal_entry_single_line_raises():
with pytest.raises(val.ValidationError) as exc:
val.validate_journal_entry(
{"id": "x"},
[{"account_id": "a", "amount": 100}],
)
assert "at least 2 lines" in str(exc.value)
def test_validate_journal_entry_zero_amount_raises():
with pytest.raises(val.ValidationError) as exc:
val.validate_journal_entry(
{"id": "x"},
[
{"account_id": "a", "amount": 0},
{"account_id": "b", "amount": 0},
],
)
assert "amount = 0" in str(exc.value)
def test_validate_journal_entry_missing_account_id_raises():
with pytest.raises(val.ValidationError) as exc:
val.validate_journal_entry(
{"id": "x"},
[
{"amount": 100},
{"account_id": "b", "amount": -100},
],
)
assert "missing account_id" in str(exc.value)
# ---------------------------------------------------------------------------
# core.validation — validate_balance
# ---------------------------------------------------------------------------
def test_validate_balance_exact_match_passes():
val.validate_balance("acct", expected_balance_sats=1000, actual_balance_sats=1000)
def test_validate_balance_within_tolerance_passes():
val.validate_balance(
"acct", expected_balance_sats=1000, actual_balance_sats=1005, tolerance_sats=10,
)
def test_validate_balance_outside_tolerance_raises():
with pytest.raises(val.ValidationError) as exc:
val.validate_balance(
"acct", expected_balance_sats=1000, actual_balance_sats=1100, tolerance_sats=10,
)
assert "Balance assertion failed" in str(exc.value)
def test_validate_balance_fiat_mismatch_raises():
with pytest.raises(val.ValidationError) as exc:
val.validate_balance(
"acct",
expected_balance_sats=1000,
actual_balance_sats=1000,
expected_balance_fiat=Decimal("100.00"),
actual_balance_fiat=Decimal("99.50"),
tolerance_fiat=Decimal("0.10"),
fiat_currency="EUR",
)
assert "Fiat balance" in str(exc.value)
# ---------------------------------------------------------------------------
# core.validation — entry-specific validators
# ---------------------------------------------------------------------------
def test_validate_receivable_entry_positive_revenue_passes():
val.validate_receivable_entry("u", amount=100, revenue_account_type="revenue")
def test_validate_receivable_entry_zero_amount_raises():
with pytest.raises(val.ValidationError):
val.validate_receivable_entry("u", amount=0, revenue_account_type="revenue")
def test_validate_receivable_entry_wrong_account_type_raises():
with pytest.raises(val.ValidationError) as exc:
val.validate_receivable_entry("u", amount=100, revenue_account_type="expense")
assert "revenue account" in str(exc.value)
def test_validate_expense_entry_non_equity_requires_expense_account():
with pytest.raises(val.ValidationError) as exc:
val.validate_expense_entry(
"u", amount=100, expense_account_type="asset", is_equity=False,
)
assert "expense account" in str(exc.value)
def test_validate_expense_entry_equity_allows_non_expense_account():
"""Equity contributions bypass the expense-account requirement."""
val.validate_expense_entry(
"u", amount=100, expense_account_type="equity", is_equity=True,
)
def test_validate_payment_entry_negative_raises():
with pytest.raises(val.ValidationError):
val.validate_payment_entry("u", amount=-1)
# ---------------------------------------------------------------------------
# core.validation — validate_metadata
# ---------------------------------------------------------------------------
def test_validate_metadata_required_keys_missing_raises():
with pytest.raises(val.ValidationError) as exc:
val.validate_metadata({"foo": 1}, required_keys=["bar", "baz"])
assert "bar" in str(exc.value) and "baz" in str(exc.value)
def test_validate_metadata_fiat_currency_without_amount_raises():
with pytest.raises(val.ValidationError) as exc:
val.validate_metadata({"fiat_currency": "EUR"})
assert "both be present or both absent" in str(exc.value)
def test_validate_metadata_fiat_amount_without_currency_raises():
with pytest.raises(val.ValidationError):
val.validate_metadata({"fiat_amount": "10.00"})
@pytest.mark.xfail(
reason="libra/issues/38 — except clause doesn't catch decimal.InvalidOperation, "
"so the raw exception leaks instead of becoming ValidationError. Flip when fixed.",
strict=True,
)
def test_validate_metadata_fiat_amount_invalid_decimal_raises():
with pytest.raises(val.ValidationError) as exc:
val.validate_metadata({"fiat_amount": "not-a-number", "fiat_currency": "EUR"})
assert "Invalid fiat_amount" in str(exc.value)
def test_validate_metadata_both_present_passes():
val.validate_metadata({"fiat_amount": "100.50", "fiat_currency": "EUR"})
def test_validate_metadata_neither_present_passes():
val.validate_metadata({"source": "api"})

212
tests/test_void_reject_api.py
View file

@ -1,212 +0,0 @@
"""Reject / void pending entry flow — `POST /libra/api/v1/entries/{id}/reject`.
Captures the current (pre-issue #24) in-place mutation behaviour:
- Pending entries (`!` flag) can be rejected by a super user.
- Rejection appends `#voided` to the transaction line in the .beancount file
(no new transaction posted — this is the only in-place edit path in libra).
- Voided entries are filtered out of balance queries.
- The reject endpoint only matches pending entries; cleared (`*`) ones return
404 because the search loop filters by `flag == '!'`.
PR #34 changes whether the user's `/entries/user` listing surfaces voided rows.
The test `test_voided_entry_excluded_from_user_journal` documents the current
("filtered") behaviour; flip it if/when that change lands.
When the reversing-entry refactor in issue #24 ships, these tests will need to
move from "void via tag append" to "void via reversal transaction." The shape
of the tests should still hold — what changes is the on-disk evidence.
"""
from uuid import uuid4
import pytest
from .helpers import (
approve_entry,
get_balance,
list_user_entries,
post_expense,
reject_entry,
)
@pytest.mark.anyio
async def test_admin_can_reject_pending_expense(
client, super_user_headers, configured_user, standard_accounts,
):
"""Happy path: user submits expense → admin rejects → response includes
the entry id, balance still zero."""
_, wallet = configured_user
posted = await post_expense(
client,
wallet_inkey=wallet.inkey,
user_wallet_id=wallet.id,
amount="15.00",
currency="EUR",
description=f"Reject me {uuid4().hex[:6]}",
expense_account=standard_accounts["expense_food"]["name"],
)
result = await reject_entry(
client, super_user_headers=super_user_headers, entry_id=posted["id"],
)
assert result.get("entry_id") == posted["id"]
balance = await get_balance(client, wallet_inkey=wallet.inkey)
assert not balance.get("fiat_balances"), (
f"voided entry should not surface in balance, got {balance}"
)
@pytest.mark.anyio
async def test_voided_entry_visible_in_user_journal(
client, super_user_headers, configured_user, standard_accounts,
):
"""Post-commit-1c89e69 behaviour: rejected entries remain visible in
the user's `/entries/user` listing so the user can see their own
rejected history rather than having it silently disappear.
The UI is expected to render these with a 'voided' visual marker
(PR #34 webapp companion). The balance query still excludes them
via the separate `tags` filter — covered in
`test_admin_can_reject_pending_expense`.
"""
_, wallet = configured_user
tag = f"void-marker-{uuid4().hex[:6]}"
posted = await post_expense(
client,
wallet_inkey=wallet.inkey,
user_wallet_id=wallet.id,
amount="20.00",
currency="EUR",
description=tag,
expense_account=standard_accounts["expense_food"]["name"],
)
await reject_entry(
client, super_user_headers=super_user_headers, entry_id=posted["id"],
)
listing = await list_user_entries(client, wallet_inkey=wallet.inkey)
entries = listing.get("entries", [])
descriptions = [e.get("description") or "" for e in entries]
assert any(tag in d for d in descriptions), (
f"voided entry should remain visible in user journal post-#34, "
f"got descriptions: {descriptions}"
)
voided = next(
(e for e in entries if tag in (e.get("description") or "")), None,
)
assert voided is not None
assert "voided" in voided.get("tags", []), (
f"voided entry should be tagged 'voided' for UI styling, "
f"got tags: {voided.get('tags')}"
)
@pytest.mark.anyio
async def test_reject_unknown_entry_returns_404(
client, super_user_headers,
):
"""An entry id that doesn't exist anywhere in the ledger 404s."""
bogus_id = uuid4().hex[:16]
r = await client.post(
f"/libra/api/v1/entries/{bogus_id}/reject",
headers=super_user_headers,
)
assert r.status_code == 404, f"expected 404, got {r.status_code}: {r.text}"
assert "not found" in r.text.lower()
@pytest.mark.anyio
async def test_reject_already_cleared_entry_returns_404(
client, super_user_headers, configured_user, standard_accounts,
):
"""The reject lookup filters by `flag == '!'` so already-approved
(cleared) entries are indistinguishable from non-existent ones —
both 404."""
_, wallet = configured_user
posted = await post_expense(
client,
wallet_inkey=wallet.inkey,
user_wallet_id=wallet.id,
amount="11.00",
currency="EUR",
description=f"Approve-then-reject {uuid4().hex[:6]}",
expense_account=standard_accounts["expense_food"]["name"],
)
await approve_entry(
client, super_user_headers=super_user_headers, entry_id=posted["id"],
)
r = await client.post(
f"/libra/api/v1/entries/{posted['id']}/reject",
headers=super_user_headers,
)
assert r.status_code == 404, f"expected 404, got {r.status_code}: {r.text}"
@pytest.mark.anyio
async def test_non_super_user_cannot_reject(
client, configured_user, standard_accounts,
):
"""Reject endpoint uses libra's `require_super_user` — wallet
admin-key of a non-super user is forbidden."""
_, wallet = configured_user
posted = await post_expense(
client,
wallet_inkey=wallet.inkey,
user_wallet_id=wallet.id,
amount="13.00",
currency="EUR",
description=f"Forbidden reject {uuid4().hex[:6]}",
expense_account=standard_accounts["expense_food"]["name"],
)
r = await client.post(
f"/libra/api/v1/entries/{posted['id']}/reject",
headers={"X-Api-Key": wallet.adminkey, "Content-type": "application/json"},
)
assert r.status_code == 403, f"expected 403, got {r.status_code}: {r.text}"
assert "super" in r.text.lower()
@pytest.mark.anyio
async def test_double_reject_returns_404_on_second_call(
client, super_user_headers, configured_user, standard_accounts,
):
"""After a successful reject the entry is no longer matched by the
lookup (it's still flag `!` but its journal-listing-filter behaviour
is "voided"). A second reject 404s rather than mutating again.
Documents the de-facto idempotency story: it's "first wins, repeat
fails cleanly" rather than "repeat is a no-op success." If the
reversing-entry refactor (#24) reshapes this, the test will reveal it.
"""
_, wallet = configured_user
posted = await post_expense(
client,
wallet_inkey=wallet.inkey,
user_wallet_id=wallet.id,
amount="9.00",
currency="EUR",
description=f"Double reject {uuid4().hex[:6]}",
expense_account=standard_accounts["expense_food"]["name"],
)
await reject_entry(
client, super_user_headers=super_user_headers, entry_id=posted["id"],
)
r = await client.post(
f"/libra/api/v1/entries/{posted['id']}/reject",
headers=super_user_headers,
)
# First reject succeeded; second reject either 404 (entry still flag !
# but matched-by-tag elsewhere) or 200 with idempotent no-op. Lock in
# whichever the current code does so a future change to the reject
# path forces a deliberate decision.
assert r.status_code in (200, 404), (
f"second reject should be deterministic, got {r.status_code}: {r.text}"
)

112
views_api.py
View file

@ -1609,8 +1609,7 @@ async def api_get_my_balance(
return UserBalance( return UserBalance(
user_id=wallet.wallet.user, user_id=wallet.wallet.user,
balance=balance_data["balance"], balance=balance_data["balance"],
accounts=[], accounts=[], # Could populate from balance_data["accounts"] if needed
account_balances=balance_data.get("accounts", []),
fiat_balances=balance_data["fiat_balances"], fiat_balances=balance_data["fiat_balances"],
total_expenses_sats=totals["total_expenses_sats"], total_expenses_sats=totals["total_expenses_sats"],
total_expenses_fiat=totals["total_expenses_fiat"], total_expenses_fiat=totals["total_expenses_fiat"],
@ -1992,11 +1991,7 @@ async def api_settle_receivable(
# DR Cash/Bank (asset increased), CR Accounts Receivable (asset decreased) # DR Cash/Bank (asset increased), CR Accounts Receivable (asset decreased)
# This records that user paid their debt # This records that user paid their debt
from .fava_client import get_fava_client from .fava_client import get_fava_client
from .beancount_format import ( from .beancount_format import format_payment_entry, format_fiat_settlement_entry
format_payment_entry,
format_fiat_settlement_entry,
format_fiat_net_settlement_entry,
)
from decimal import Decimal from decimal import Decimal
fava = get_fava_client() fava = get_fava_client()
@ -2006,106 +2001,9 @@ async def api_settle_receivable(
"cash", "bank_transfer", "check", "other" "cash", "bank_transfer", "check", "other"
] ]
if is_fiat_payment and data.settled_entry_links is None: if is_fiat_payment:
# Auto-detect netting + credit-overflow path (libra-#33 + libra-#41). # Fiat currency payment (cash, bank transfer, etc.)
# The operator hasn't picked specific entries — backend nets all # Record in fiat currency with sats as metadata
# open balances in both directions, validates cash matches the net
# obligation (or absorbs excess into credit), and writes a single
# transaction that links every reconciled source entry.
unsettled_payables = await fava.get_unsettled_entries_bql(data.user_id, "expense")
unsettled_receivables = await fava.get_unsettled_entries_bql(data.user_id, "receivable")
payable_total = sum(
(Decimal(str(e["fiat_amount"])) for e in unsettled_payables),
Decimal(0),
)
receivable_total = sum(
(Decimal(str(e["fiat_amount"])) for e in unsettled_receivables),
Decimal(0),
)
all_links = (
[e["link"] for e in unsettled_payables if e.get("link")]
+ [e["link"] for e in unsettled_receivables if e.get("link")]
)
if receivable_total <= 0:
# Endpoint is `/receivables/settle` — user paying off something
# they owe. With no open receivable there's nothing this endpoint
# can settle. Operator should use `/payables/pay` (libra pays user)
# or wait until the user has open receivables.
raise HTTPException(
status_code=HTTPStatus.BAD_REQUEST,
detail=(
f"User {data.user_id[:8]} has no open receivables to settle. "
f"If libra owes them, use `/payables/pay`. If they want to "
f"deposit credit without an open obligation, that's a future "
f"feature (libra-#41 follow-up)."
),
)
cash_paid = Decimal(str(data.amount))
net_obligation = receivable_total - payable_total
tolerance = Decimal("0.01") # forex rounding slack
if cash_paid + tolerance < net_obligation:
# Under-pay without explicit entry-picks — backend can't guess
# which receivable(s) the operator means to settle.
raise HTTPException(
status_code=HTTPStatus.BAD_REQUEST,
detail={
"message": (
"Cash paid is less than net obligation. Pay the exact "
"net to clear all open entries, or pass "
"`settled_entry_links` to settle a specific subset."
),
"cash_paid": float(cash_paid),
"net_obligation": float(net_obligation),
"receivable_total": float(receivable_total),
"payable_total": float(payable_total),
"currency": data.currency.upper(),
},
)
credit_overflow = cash_paid - net_obligation
if credit_overflow < tolerance:
credit_overflow = Decimal(0)
# Auto-create the user-side accounts as needed.
user_payable = None
if payable_total > 0:
user_payable = await get_or_create_user_account(
data.user_id, AccountType.LIABILITY, "Accounts Payable",
)
user_credit = None
if credit_overflow > 0:
user_credit = await get_or_create_user_account(
data.user_id, AccountType.LIABILITY, "Credit",
)
entry = format_fiat_net_settlement_entry(
user_id=data.user_id,
cash_account=payment_account.name,
receivable_account=user_receivable.name,
payable_account=user_payable.name if user_payable else None,
credit_account=user_credit.name if user_credit else None,
cash_paid_fiat=cash_paid,
total_receivable_fiat=receivable_total,
total_payable_fiat=payable_total,
credit_overflow_fiat=credit_overflow,
fiat_currency=data.currency.upper(),
description=data.description,
entry_date=datetime.now().date(),
payment_method=data.payment_method,
reference=data.reference or f"MANUAL-{data.user_id[:8]}",
settled_entry_links=all_links,
)
elif is_fiat_payment:
# Legacy fiat path — operator provided `settled_entry_links` explicitly,
# meaning they're settling a specific subset. Backwards-compatible
# 2-leg behaviour: trust the caller's list, no auto-netting, no
# credit-overflow validation. Use the auto-detect path above (omit
# settled_entry_links) to get netting + credit handling.
if not data.amount_sats: if not data.amount_sats:
raise HTTPException( raise HTTPException(
status_code=HTTPStatus.BAD_REQUEST, status_code=HTTPStatus.BAD_REQUEST,