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

Add integration test suite

Browse source 
113 passing tests + 3 skipped + 8 xfailed across 10 files, covering user
expense and income flow, admin receivable/revenue, settings + auth gates,
void/reject, manual payment requests, balance display, Lightning auth
paths, reconciliation API, and pure-function units. Runs against a real
Fava subprocess and full LNbits app via asgi_lifespan; the harness
captures the auth-flow / settings / env-var disciplines surfaced during
build-out (see tests/README.md and tests/conftest.py docstring).

Eight xfailed/skipped tests carry full implementations gated behind issues
#38, #39, #40 — they flip back on automatically when those land.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Padreug 2026-06-07 09:47:09 +02:00
commit 7a4b3022c2
14 changed files with 3710 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

686
tests/conftest.py Normal file
View file

@ -0,0 +1,686 @@
"""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