Michilis/lnbits
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test: some tests

Browse Source
This commit is contained in:
Vlad Stan
2026-03-25 10:49:52 +02:00
parent 2cce687865
commit 3c77acf266
9 changed files with 1177 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/unit/test_cache.py
+56 -1
View File
@@ -1,8 +1,11 @@
import asyncio
from time import time
import pytest
from pytest_mock.plugin import MockerFixture
from lnbits.utils.cache import Cache
from lnbits.settings import Settings
from lnbits.utils.cache import Cache, Cached
key = "foo"
value = "bar"
@@ -59,3 +62,55 @@ async def test_cache_coro(cache):
await cache.save_result(test, key="test")
result = await cache.save_result(test, key="test")
assert result == called == 1
def test_cached_older_than():
cached = Cached(value="value", expiry=time() - 5)
assert cached.older_than(1) is True
assert cached.older_than(10) is False
@pytest.mark.anyio
async def test_cache_value_returns_cached_metadata(cache):
cache.set(key, value, expiry=1)
cached = cache.value(key)
assert cached is not None
assert cached.value == value
assert cached.expiry > time()
@pytest.mark.anyio
async def test_cache_pop_expired_returns_default(cache):
cache.set(key, value, expiry=0.01)
await asyncio.sleep(0.02)
assert cache.pop(key, default="fallback") == "fallback"
@pytest.mark.anyio
async def test_invalidate_forever_logs_and_recovers_from_errors(
settings: Settings, mocker: MockerFixture
):
test_cache = Cache(interval=0)
logger_error = mocker.patch("lnbits.utils.cache.logger.error")
original_running = settings.lnbits_running
calls = 0
async def fake_sleep(_interval):
nonlocal calls
calls += 1
if calls == 1:
raise RuntimeError("boom")
settings.lnbits_running = False
try:
settings.lnbits_running = True
mocker.patch("lnbits.utils.cache.asyncio.sleep", side_effect=fake_sleep)
await test_cache.invalidate_forever()
finally:
settings.lnbits_running = original_running
logger_error.assert_called_once_with("Error invalidating cache")
tests/unit/test_crypto_aes.py
+91 -2
View File
@@ -1,6 +1,15 @@
import pytest
from base64 import b64encode
from hashlib import sha256
from lnbits.utils.crypto import AESCipher
import pytest
from pytest_mock.plugin import MockerFixture
from lnbits.utils.crypto import (
AESCipher,
fake_privkey,
random_secret_and_hash,
verify_preimage,
)
@pytest.mark.anyio
@@ -18,3 +27,83 @@ async def test_aes_encrypt_decrypt(key):
encrypted_text = aes.encrypt(original_text.encode())
decrypted_text = aes.decrypt(encrypted_text)
assert original_text == decrypted_text
def test_random_secret_and_hash():
secret, payment_hash = random_secret_and_hash(16)
assert len(secret) == 32
assert payment_hash == sha256(bytes.fromhex(secret)).hexdigest()
def test_fake_privkey_is_deterministic():
assert fake_privkey("secret") == fake_privkey("secret")
assert fake_privkey("secret") != fake_privkey("other-secret")
def test_verify_preimage_success_and_failure():
preimage = "00" * 32
payment_hash = sha256(bytes.fromhex(preimage)).hexdigest()
assert verify_preimage(preimage, payment_hash) is True
assert verify_preimage(preimage, "0" * 64) is False
@pytest.mark.anyio
async def test_aes_urlsafe_encrypt_decrypt():
aes = AESCipher("normal_string")
encrypted_text = aes.encrypt(b"url-safe", urlsafe=True)
assert aes.decrypt(encrypted_text, urlsafe=True) == "url-safe"
def test_aes_derive_iv_and_key_requires_eight_byte_salt():
aes = AESCipher("normal_string")
with pytest.raises(ValueError, match="Salt must be 8 bytes"):
aes.derive_iv_and_key(b"short")
def test_aes_decrypt_rejects_invalid_salt_prefix():
aes = AESCipher("normal_string")
encrypted_text = b64encode(b"NotSalted__12345678ciphertext").decode()
with pytest.raises(ValueError, match="Invalid salt."):
aes.decrypt(encrypted_text)
def test_aes_decrypt_raises_for_cipher_errors(mocker: MockerFixture):
aes = AESCipher("normal_string")
fake_cipher = mocker.Mock()
fake_cipher.decrypt.side_effect = RuntimeError("boom")
mocker.patch.object(
aes,
"derive_iv_and_key",
return_value=(b"0" * aes.block_size, b"1" * 32),
)
mocker.patch("lnbits.utils.crypto.AES.new", return_value=fake_cipher)
encrypted_text = b64encode(b"Salted__12345678ciphertext").decode()
with pytest.raises(ValueError, match="Could not decrypt payload"):
aes.decrypt(encrypted_text)
def test_aes_decrypt_raises_for_invalid_utf8_output(mocker: MockerFixture):
aes = AESCipher("normal_string")
fake_cipher = mocker.Mock()
fake_cipher.decrypt.return_value = b"\xff\x01"
mocker.patch.object(
aes,
"derive_iv_and_key",
return_value=(b"0" * aes.block_size, b"1" * 32),
)
mocker.patch("lnbits.utils.crypto.AES.new", return_value=fake_cipher)
encrypted_text = b64encode(b"Salted__12345678ciphertext").decode()
with pytest.raises(ValueError, match="invalid UTF-8 data"):
aes.decrypt(encrypted_text)
tests/unit/test_decorators.py
+90 -1
View File
@@ -11,7 +11,16 @@ from pydantic.types import UUID4
from lnbits.core.crud.users import delete_account
from lnbits.core.models import User
from lnbits.core.models.users import AccessTokenPayload
from lnbits.decorators import check_user_exists
from lnbits.decorators import (
access_token_payload,
check_access_token,
check_admin_ui,
check_extension_builder,
check_first_install,
check_user_exists,
optional_user_id,
)
from lnbits.helpers import create_access_token
from lnbits.settings import AuthMethods, Settings, settings
@@ -136,3 +145,83 @@ async def test_check_user_exists_with_user_id_only_not_allowed(user_alan: User):
await check_user_exists(request, access_token=None, usr=UUID4(user_alan.id))
assert exc_info.value.status_code == 401
assert exc_info.value.detail == "Missing user ID or access token."
@pytest.mark.anyio
async def test_check_access_token_prefers_available_source():
assert await check_access_token("header", "cookie", "bearer") == "header"
assert await check_access_token(None, "cookie", "bearer") == "cookie"
assert await check_access_token(None, None, "bearer") == "bearer"
@pytest.mark.anyio
async def test_access_token_payload_success_and_missing(settings: Settings):
token = create_access_token({"sub": "alice", "usr": "user-id"})
payload = await access_token_payload(token)
assert isinstance(payload, AccessTokenPayload)
assert payload.sub == "alice"
assert payload.usr == "user-id"
with pytest.raises(HTTPException, match="Missing access token."):
await access_token_payload(None)
@pytest.mark.anyio
async def test_optional_user_id_uses_user_id_or_access_token(
user_alan: User, settings: Settings
):
settings.auth_allowed_methods = [AuthMethods.user_id_only.value]
request = Request({"type": "http", "path": "/wallet", "method": "GET"})
assert (
await optional_user_id(request, access_token=None, usr=UUID4(user_alan.id))
== user_alan.id
)
settings.auth_allowed_methods = []
token = create_access_token({"sub": user_alan.username, "usr": user_alan.id})
assert await optional_user_id(request, access_token=token, usr=None) == user_alan.id
assert await optional_user_id(request, access_token=None, usr=None) is None
@pytest.mark.anyio
async def test_check_admin_ui_and_first_install(settings: Settings):
original_admin_ui = settings.lnbits_admin_ui
original_first_install = settings.first_install
try:
settings.lnbits_admin_ui = False
with pytest.raises(HTTPException, match="Admin UI is disabled."):
await check_admin_ui()
settings.lnbits_admin_ui = True
await check_admin_ui()
settings.first_install = False
with pytest.raises(
HTTPException, match="Super user account has already been configured."
):
await check_first_install()
settings.first_install = True
await check_first_install()
finally:
settings.lnbits_admin_ui = original_admin_ui
settings.first_install = original_first_install
@pytest.mark.anyio
async def test_check_extension_builder_requires_admin_when_disabled_for_users(
settings: Settings, user_alan: User
):
settings.lnbits_extensions_builder_activate_non_admins = False
with pytest.raises(
HTTPException, match="Extension Builder is disabled for non admin users."
):
await check_extension_builder(user_alan)
admin_user = user_alan.copy(deep=True)
admin_user.admin = True
await check_extension_builder(admin_user)
tests/unit/test_exchange_rates.py
+203
View File
@@ -1,8 +1,52 @@
from unittest.mock import AsyncMock
import httpx
import pytest
from pytest_mock.plugin import MockerFixture
from lnbits.settings import ExchangeRateProvider, Settings
from lnbits.utils.exchange_rates import (
allowed_currencies,
apply_trimmed_mean_filter,
btc_price,
btc_rates,
fiat_amount_as_satoshis,
get_fiat_rate_and_price_satoshis,
get_fiat_rate_satoshis,
satoshis_amount_as_fiat,
)
class MockResponse:
def __init__(self, *, text: str = "", json_data=None, error: Exception | None = None):
self.text = text
self._json_data = json_data or {}
self._error = error
def raise_for_status(self):
if self._error:
raise self._error
def json(self):
return self._json_data
class MockAsyncClient:
def __init__(self, response: MockResponse):
self.response = response
self.calls: list[tuple[str, int]] = []
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc, tb):
return False
async def get(self, url: str, timeout: int = 3):
self.calls.append((url, timeout))
return self.response
class TestApplyTrimmedMeanFilter:
"""Test the trimmed mean filtering function"""
@@ -123,3 +167,162 @@ class TestApplyTrimmedMeanFilter:
# Should keep the rate at exactly 1% deviation
assert len(result) == 2
assert result == rates
def test_allowed_currencies_returns_full_list_by_default(settings: Settings):
original_allowed_currencies = settings.lnbits_allowed_currencies
try:
settings.lnbits_allowed_currencies = []
currencies = allowed_currencies()
assert "USD" in currencies
assert "EUR" in currencies
finally:
settings.lnbits_allowed_currencies = original_allowed_currencies
def test_allowed_currencies_respects_allow_list(settings: Settings):
original_allowed_currencies = settings.lnbits_allowed_currencies
try:
settings.lnbits_allowed_currencies = ["USD", "EUR"]
assert allowed_currencies() == ["EUR", "USD"]
finally:
settings.lnbits_allowed_currencies = original_allowed_currencies
@pytest.mark.anyio
async def test_btc_rates_rejects_disallowed_currency(settings: Settings):
original_allowed_currencies = settings.lnbits_allowed_currencies
try:
settings.lnbits_allowed_currencies = ["EUR"]
with pytest.raises(ValueError, match="Currency 'usd' not allowed."):
await btc_rates("usd")
finally:
settings.lnbits_allowed_currencies = original_allowed_currencies
@pytest.mark.anyio
async def test_btc_rates_parses_plain_text_response(
settings: Settings, mocker: MockerFixture
):
provider = ExchangeRateProvider(
name="PlainText",
api_url="https://plain.test/{TO}",
path="",
)
client = MockAsyncClient(MockResponse(text="12,345.67"))
mocker.patch.object(settings, "lnbits_allowed_currencies", [])
mocker.patch.object(settings, "lnbits_exchange_rate_providers", [provider])
mocker.patch("lnbits.utils.exchange_rates.httpx.AsyncClient", return_value=client)
rates = await btc_rates("usd")
assert rates == [("PlainText", 12345.67)]
assert client.calls == [("https://plain.test/USD", 3)]
@pytest.mark.anyio
async def test_btc_rates_parses_json_path_response(
settings: Settings, mocker: MockerFixture
):
provider = ExchangeRateProvider(
name="JsonProvider",
api_url="https://json.test/{TO}",
path="$.data.rates.{TO}",
)
client = MockAsyncClient(
MockResponse(json_data={"data": {"rates": {"USD": "54321.0"}}})
)
mocker.patch.object(settings, "lnbits_allowed_currencies", [])
mocker.patch.object(settings, "lnbits_exchange_rate_providers", [provider])
mocker.patch("lnbits.utils.exchange_rates.httpx.AsyncClient", return_value=client)
rates = await btc_rates("usd")
assert rates == [("JsonProvider", 54321.0)]
assert client.calls == [("https://json.test/USD", 3)]
@pytest.mark.anyio
async def test_btc_rates_skips_unsupported_and_failing_providers(
settings: Settings, mocker: MockerFixture
):
unsupported = ExchangeRateProvider(
name="Unsupported",
api_url="https://unsupported.test/{TO}",
path="$.price",
exclude_to=["usd"],
)
failing = ExchangeRateProvider(
name="Failing",
api_url="https://failing.test/{TO}",
path="$.price",
)
client = MockAsyncClient(MockResponse(error=httpx.HTTPError("boom")))
mocker.patch.object(settings, "lnbits_allowed_currencies", [])
mocker.patch.object(settings, "lnbits_exchange_rate_providers", [unsupported, failing])
mocker.patch("lnbits.utils.exchange_rates.httpx.AsyncClient", return_value=client)
assert await btc_rates("usd") == []
@pytest.mark.anyio
async def test_btc_price_handles_empty_single_and_multiple_rates(mocker: MockerFixture):
mocker.patch("lnbits.utils.exchange_rates.btc_rates", AsyncMock(return_value=[]))
assert await btc_price("usd") == 0.0
mocker.patch(
"lnbits.utils.exchange_rates.btc_rates",
AsyncMock(return_value=[("Only", 50000.0)]),
)
assert await btc_price("usd") == 50000.0
mocker.patch(
"lnbits.utils.exchange_rates.btc_rates",
AsyncMock(return_value=[("A", 40000.0), ("B", 50000.0)]),
)
assert await btc_price("usd") == 45000.0
@pytest.mark.anyio
async def test_rate_and_amount_conversion_helpers(mocker: MockerFixture):
cache_result = AsyncMock(return_value=50000.0)
mocker.patch("lnbits.utils.exchange_rates.cache.save_result", cache_result)
rate, price = await get_fiat_rate_and_price_satoshis("usd")
assert price == 50000.0
assert rate == 2000.0
cache_result.assert_awaited_once()
mocker.patch(
"lnbits.utils.exchange_rates.get_fiat_rate_and_price_satoshis",
AsyncMock(return_value=(1250.0, 80000.0)),
)
assert await get_fiat_rate_satoshis("usd") == 1250.0
mocker.patch(
"lnbits.utils.exchange_rates.get_fiat_rate_satoshis",
AsyncMock(return_value=100.0),
)
assert await fiat_amount_as_satoshis(2.5, "usd") == 250
assert await satoshis_amount_as_fiat(500, "usd") == 5.0
@pytest.mark.anyio
async def test_amount_conversion_helpers_raise_when_rate_missing(
mocker: MockerFixture,
):
mocker.patch(
"lnbits.utils.exchange_rates.get_fiat_rate_satoshis",
AsyncMock(return_value=0.0),
)
with pytest.raises(ValueError, match="Could not get exchange rate for usd."):
await fiat_amount_as_satoshis(1, "usd")
with pytest.raises(ValueError, match="Could not get exchange rate for usd."):
await satoshis_amount_as_fiat(100, "usd")
tests/unit/test_fiat_providers.py
+230 -2
View File
@@ -9,20 +9,63 @@ from pytest_mock.plugin import MockerFixture
from lnbits.core.crud.payments import get_payments
from lnbits.core.crud.users import get_user
from lnbits.core.crud.wallets import create_wallet
from lnbits.core.models.payments import CreateInvoice, PaymentState
from lnbits.core.models.payments import CreateInvoice, Payment, PaymentState
from lnbits.core.models.users import User
from lnbits.core.models.wallets import Wallet
from lnbits.core.services import check_payment_status, payments
from lnbits.core.services.fiat_providers import (
check_fiat_status,
check_stripe_signature,
handle_fiat_payment_confirmation,
test_connection as fiat_provider_connection,
verify_paypal_webhook,
)
from lnbits.core.services.users import create_user_account
from lnbits.fiat.base import FiatInvoiceResponse, FiatPaymentStatus
from lnbits.fiat.base import FiatInvoiceResponse, FiatPaymentStatus, FiatStatusResponse
from lnbits.settings import Settings
from tests.helpers import get_random_string
class MockHTTPResponse:
def __init__(self, json_data=None, error: Exception | None = None):
self._json_data = json_data or {}
self._error = error
def raise_for_status(self):
if self._error:
raise self._error
def json(self):
return self._json_data
class MockHTTPClient:
def __init__(self, responses: list[MockHTTPResponse]):
self._responses = responses
self.calls: list[tuple[str, dict]] = []
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc, tb):
return False
async def post(self, path: str, **kwargs):
self.calls.append((path, kwargs))
return self._responses.pop(0)
@pytest.fixture(autouse=True)
def fiat_provider_test_settings(settings: Settings):
original_allowed_currencies = settings.lnbits_allowed_currencies
original_paypal_enabled = settings.paypal_enabled
settings.lnbits_allowed_currencies = []
settings.paypal_enabled = False
yield
settings.lnbits_allowed_currencies = original_allowed_currencies
settings.paypal_enabled = original_paypal_enabled
@pytest.mark.anyio
async def test_create_wallet_fiat_invoice_missing_provider():
invoice_data = CreateInvoice(
@@ -433,3 +476,188 @@ def _make_stripe_sig_header(payload, secret, timestamp=None):
secret.encode(), signed_payload.encode(), hashlib.sha256
).hexdigest()
return f"t={timestamp},v1={signature}", timestamp, signature
@pytest.mark.anyio
async def test_check_fiat_status_handles_internal_states(mocker: MockerFixture):
pending_payment = Payment(
checking_id="external_payment",
payment_hash="hash_pending",
wallet_id="wallet_id",
amount=1000,
fee=0,
bolt11="bolt11",
status=PaymentState.PENDING,
)
success_payment = Payment(
checking_id="fiat_success",
payment_hash="hash_success",
wallet_id="wallet_id",
amount=1000,
fee=0,
bolt11="bolt11",
status=PaymentState.SUCCESS,
fiat_provider="stripe",
)
failed_payment = Payment(
checking_id="fiat_failed",
payment_hash="hash_failed",
wallet_id="wallet_id",
amount=1000,
fee=0,
bolt11="bolt11",
status=PaymentState.FAILED,
fiat_provider="stripe",
)
assert (await check_fiat_status(pending_payment)).pending is True
assert (await check_fiat_status(success_payment)).success is True
assert (await check_fiat_status(failed_payment)).failed is True
provider = mocker.Mock()
provider.get_invoice_status = AsyncMock(return_value=FiatPaymentStatus(paid=True))
mocker.patch(
"lnbits.core.services.fiat_providers.get_fiat_provider",
AsyncMock(return_value=provider),
)
queue_put = mocker.patch("lnbits.tasks.internal_invoice_queue.put", AsyncMock())
success_status = await check_fiat_status(
Payment(
checking_id="fiat_pending",
payment_hash="hash_queue",
wallet_id="wallet_id",
amount=1000,
fee=0,
bolt11="bolt11",
status=PaymentState.PENDING,
fiat_provider="stripe",
extra={"fiat_checking_id": "stripe_checking_id"},
)
)
assert success_status.success is True
queue_put.assert_awaited_once_with("fiat_pending")
await check_fiat_status(
Payment(
checking_id="fiat_pending_skip",
payment_hash="hash_skip",
wallet_id="wallet_id",
amount=1000,
fee=0,
bolt11="bolt11",
status=PaymentState.PENDING,
fiat_provider="stripe",
extra={"fiat_checking_id": "stripe_checking_id"},
),
skip_internal_payment_notifications=True,
)
assert queue_put.await_count == 1
@pytest.mark.anyio
async def test_verify_paypal_webhook_requires_configuration(settings: Settings):
settings.paypal_webhook_id = None
with pytest.raises(
ValueError, match="PayPal webhook cannot be verified. Missing webhook ID."
):
await verify_paypal_webhook({}, b"{}")
@pytest.mark.anyio
async def test_verify_paypal_webhook_requires_headers(settings: Settings):
settings.paypal_webhook_id = "webhook-id"
with pytest.raises(
ValueError, match="PayPal webhook cannot be verified. Missing headers."
):
await verify_paypal_webhook({}, b"{}")
@pytest.mark.anyio
async def test_verify_paypal_webhook_success(
settings: Settings, mocker: MockerFixture
):
settings.paypal_webhook_id = "webhook-id"
client = MockHTTPClient(
[
MockHTTPResponse(json_data={"access_token": "token"}),
MockHTTPResponse(json_data={"verification_status": "SUCCESS"}),
]
)
mocker.patch(
"lnbits.core.services.fiat_providers.httpx.AsyncClient",
return_value=client,
)
await verify_paypal_webhook(
{
"PAYPAL-TRANSMISSION-ID": "tx-id",
"PAYPAL-TRANSMISSION-TIME": "2024-01-01T00:00:00Z",
"PAYPAL-TRANSMISSION-SIG": "signature",
"PAYPAL-CERT-URL": "https://cert.example.com",
"PAYPAL-AUTH-ALGO": "SHA256withRSA",
},
b'{"id":"event-1"}',
)
assert client.calls[0][0] == "/v1/oauth2/token"
assert client.calls[1][0] == "/v1/notifications/verify-webhook-signature"
assert client.calls[1][1]["headers"]["Authorization"] == "Bearer token"
@pytest.mark.anyio
async def test_verify_paypal_webhook_raises_on_failed_verification(
settings: Settings, mocker: MockerFixture
):
settings.paypal_webhook_id = "webhook-id"
client = MockHTTPClient(
[
MockHTTPResponse(json_data={"access_token": "token"}),
MockHTTPResponse(json_data={"verification_status": "FAILURE"}),
]
)
mocker.patch(
"lnbits.core.services.fiat_providers.httpx.AsyncClient",
return_value=client,
)
with pytest.raises(ValueError, match="PayPal webhook cannot be verified."):
await verify_paypal_webhook(
{
"PAYPAL-TRANSMISSION-ID": "tx-id",
"PAYPAL-TRANSMISSION-TIME": "2024-01-01T00:00:00Z",
"PAYPAL-TRANSMISSION-SIG": "signature",
"PAYPAL-CERT-URL": "https://cert.example.com",
"PAYPAL-AUTH-ALGO": "SHA256withRSA",
},
b'{"id":"event-1"}',
)
@pytest.mark.anyio
async def test_test_connection_reports_provider_status(mocker: MockerFixture):
mocker.patch(
"lnbits.core.services.fiat_providers.get_fiat_provider",
AsyncMock(return_value=None),
)
missing_status = await fiat_provider_connection("stripe")
assert missing_status.success is False
assert missing_status.message == "Fiat provider 'stripe' not found."
provider = mocker.Mock()
provider.status = AsyncMock(return_value=FiatStatusResponse(error_message="bad key"))
mocker.patch(
"lnbits.core.services.fiat_providers.get_fiat_provider",
AsyncMock(return_value=provider),
)
error_status = await fiat_provider_connection("stripe")
assert error_status.success is False
assert error_status.message == "Cconnection test failed: bad key"
provider.status = AsyncMock(return_value=FiatStatusResponse(balance=21.0))
success_status = await fiat_provider_connection("stripe")
assert success_status.success is True
assert success_status.message == "Connection test successful. Balance: 21.0."
tests/unit/test_helpers.py
+196 -2
View File
@@ -1,6 +1,39 @@
import pytest
import hashlib
from lnbits.helpers import check_callback_url
import jwt
import pytest
from fastapi import FastAPI
from lnbits.helpers import (
camel_to_snake,
camel_to_words,
check_callback_url,
create_access_token,
decrypt_internal_message,
download_url,
encrypt_internal_message,
file_hash,
filter_dict_keys,
get_api_routes,
get_db_vendor_name,
is_camel_case,
is_lnbits_version_ok,
is_snake_case,
is_valid_email_address,
is_valid_external_id,
is_valid_label,
is_valid_pubkey,
is_valid_username,
lowercase_first_letter,
normalize_endpoint,
normalize_path,
path_segments,
sha256s,
snake_to_camel,
static_url_for,
url_for,
version_parse,
)
from lnbits.settings import Settings
@@ -82,3 +115,164 @@ def test_check_callback_url_multiple_rules(settings: Settings):
settings.lnbits_callback_url_rules.append("https://localhost:3000")
check_callback_url("https://localhost:3000/callback") # should not raise
def test_get_db_vendor_name(settings: Settings):
original_database_url = settings.lnbits_database_url
try:
settings.lnbits_database_url = None
assert get_db_vendor_name() == "SQLite"
settings.lnbits_database_url = "postgres://localhost/db"
assert get_db_vendor_name() == "PostgreSQL"
settings.lnbits_database_url = "cockroachdb://localhost/db"
assert get_db_vendor_name() == "CockroachDB"
finally:
settings.lnbits_database_url = original_database_url
def test_url_helpers(settings: Settings):
assert url_for("/api/v1/wallet", external=False, usr="user") == (
"/api/v1/wallet?usr=user&"
)
assert url_for("/api/v1/wallet", external=True, usr="user") == (
f"http://{settings.host}:{settings.port}/api/v1/wallet?usr=user&"
)
assert static_url_for("static", "bundle.min.js") == (
f"/static/bundle.min.js?v={settings.server_startup_time}"
)
@pytest.mark.parametrize(
("value", "validator"),
[
("alice@example.com", is_valid_email_address),
("alice_1", is_valid_username),
("Label 1", is_valid_label),
("external-id-1", is_valid_external_id),
("a" * 64, is_valid_pubkey),
],
)
def test_validation_helpers_valid(value, validator):
assert validator(value) is True
@pytest.mark.parametrize(
("value", "validator"),
[
("alice@example", is_valid_email_address),
("_alice", is_valid_username),
("bad/label", is_valid_label),
("contains spaces", is_valid_external_id),
("xyz", is_valid_pubkey),
],
)
def test_validation_helpers_invalid(value, validator):
assert validator(value) is False
def test_is_valid_external_id_rejects_long_and_multiline_values():
assert is_valid_external_id("x" * 257) is False
assert is_valid_external_id("evil\nnewline") is False
def test_access_token_and_internal_message_helpers(settings: Settings):
token = create_access_token(
{"sub": "alice", "usr": None, "email": "alice@example.com"},
token_expire_minutes=1,
)
payload = jwt.decode(token, settings.auth_secret_key, ["HS256"])
assert payload["sub"] == "alice"
assert payload["email"] == "alice@example.com"
assert "usr" not in payload
assert "exp" in payload
assert encrypt_internal_message(None) is None
assert decrypt_internal_message(None) is None
encrypted = encrypt_internal_message("secret-message", urlsafe=True)
assert encrypted is not None
assert decrypt_internal_message(encrypted, urlsafe=True) == "secret-message"
def test_filter_dict_keys_returns_copy_when_no_filters():
original = {"a": 1, "b": 2}
clone = filter_dict_keys(original, None)
filtered = filter_dict_keys(original, ["b", "missing"])
assert clone == original
assert clone is not original
assert filtered == {"b": 2}
def test_version_helpers(settings: Settings):
original_version = settings.version
try:
settings.version = "1.2.3"
assert version_parse("1.2.3rc4") == version_parse("1.2.3")
assert version_parse("invalid-version") == version_parse("0.0.0")
assert is_lnbits_version_ok("1.2.0", "2.0.0") is True
assert is_lnbits_version_ok("2.0.0", None) is False
assert is_lnbits_version_ok(None, "1.2.3") is False
finally:
settings.version = original_version
def test_download_url_rejects_non_http_schemes(tmp_path):
with pytest.raises(
ValueError, match="Invalid URL: ftp://example.com. Must start with 'http'"
):
download_url("ftp://example.com", tmp_path / "download.bin")
def test_file_hash(tmp_path):
filename = tmp_path / "payload.txt"
filename.write_text("hello world")
assert file_hash(filename) == hashlib.sha256(b"hello world").hexdigest()
def test_get_api_routes_extracts_v1_paths():
app = FastAPI()
@app.get("/api/v1/payments")
async def payments():
return {}
@app.get("/myext/api/v1/settings")
async def extension_settings():
return {}
@app.get("/health")
async def health():
return {}
routes = get_api_routes([*app.routes, object()])
assert routes == {
"/api/v1/payments": "Payments",
"/myext/api/v1": "Myext",
}
def test_path_and_case_helpers():
assert path_segments("/wallet/path") == ["wallet", "path"]
assert path_segments("/upgrades/ext/assets/app.js") == ["assets", "app.js"]
assert normalize_path(None) == "/"
assert normalize_path("/upgrades/ext/assets/app.js") == "/assets/app.js"
assert normalize_endpoint("example.com/") == "https://example.com"
assert normalize_endpoint("ws://socket.example.com") == "ws://socket.example.com"
assert normalize_endpoint("http://example.com/", add_proto=False) == "http://example.com"
assert camel_to_words("CamelCaseName") == "Camel Case Name"
assert camel_to_snake("CamelCaseName") == "camel_case_name"
assert snake_to_camel("snake_case_name") == "snakeCaseName"
assert snake_to_camel("snake_case_name", capitalize_first=True) == "SnakeCaseName"
assert is_camel_case("CamelCase1") is True
assert is_camel_case("camelCase") is False
assert is_snake_case("snake_case_1") is True
assert is_snake_case("SnakeCase") is False
assert lowercase_first_letter("Hello") == "hello"
assert sha256s("hello") == hashlib.sha256(b"hello").hexdigest()
tests/unit/test_helpers_query.py
+20
View File
@@ -3,6 +3,7 @@ import json
import pytest
from lnbits.db import (
dict_to_submodel,
dict_to_model,
insert_query,
model_to_dict,
@@ -83,3 +84,22 @@ async def test_helpers_dict_to_model():
assert m.active is True
assert type(m.child) is DbTestModel2
assert type(m.child.child) is DbTestModel
@pytest.mark.anyio
async def test_helpers_dict_to_submodel():
model = dict_to_submodel(
DbTestModel,
'{"id": 9, "name": "submodel", "value": "value"}',
)
assert model == DbTestModel(id=9, name="submodel", value="value")
assert dict_to_submodel(DbTestModel, "") is None
assert dict_to_submodel(DbTestModel, "null") is None
@pytest.mark.anyio
async def test_helpers_dict_to_model_ignores_unknown_fields():
model = dict_to_model({**test_dict, "ignored": "field"}, DbTestModel3)
assert model == test_data
tests/unit/test_services_settings.py
+124
View File
@@ -0,0 +1,124 @@
from unittest.mock import AsyncMock
import pytest
from pydantic import ValidationError
from pytest_mock.plugin import MockerFixture
from lnbits.core.services.settings import (
check_webpush_settings,
dict_to_settings,
update_cached_settings,
)
from lnbits.settings import Settings
class FakePublicKey:
def public_bytes(self, *_args, **_kwargs):
return b"public-bytes"
class FakeVapid:
def __init__(self, has_public_key: bool = True):
self.public_key = FakePublicKey() if has_public_key else None
def generate_keys(self):
return None
def private_pem(self):
return b"private-key"
def test_dict_to_settings_parses_known_values():
parsed = dict_to_settings(
{
"lnbits_site_title": "Test Title",
"lnbits_service_fee": 5,
"ignored_field": "ignored",
}
)
assert parsed.lnbits_site_title == "Test Title"
assert parsed.lnbits_service_fee == 5
assert not hasattr(parsed, "ignored_field")
def test_dict_to_settings_validates_invalid_values():
with pytest.raises(ValidationError):
dict_to_settings({"lnbits_service_fee": "not-a-number"})
def test_update_cached_settings_updates_runtime_values(settings: Settings):
original_title = settings.lnbits_site_title
original_host = settings.host
original_super_user = settings.super_user
try:
update_cached_settings(
{
"lnbits_site_title": "Updated",
"host": "forbidden-host",
"super_user": "super-user-id",
"missing_field": "ignored",
}
)
assert settings.lnbits_site_title == "Updated"
assert settings.host == original_host
assert settings.super_user == "super-user-id"
finally:
settings.lnbits_site_title = original_title
settings.host = original_host
settings.super_user = original_super_user
@pytest.mark.anyio
async def test_check_webpush_settings_generates_and_persists_keys(
settings: Settings, mocker: MockerFixture
):
mocker.patch.object(settings, "lnbits_webpush_privkey", "")
mocker.patch.object(settings, "lnbits_webpush_pubkey", None)
mocker.patch.object(settings, "lnbits_admin_ui", True)
mocker.patch("lnbits.core.services.settings.Vapid", return_value=FakeVapid())
mocker.patch("lnbits.core.services.settings.b64urlencode", return_value="public-key")
update_admin = mocker.patch(
"lnbits.core.services.settings.update_admin_settings",
AsyncMock(),
)
await check_webpush_settings()
assert settings.lnbits_webpush_privkey == "private-key"
assert settings.lnbits_webpush_pubkey == "public-key"
update_admin.assert_awaited_once()
@pytest.mark.anyio
async def test_check_webpush_settings_requires_public_key(
settings: Settings, mocker: MockerFixture
):
mocker.patch.object(settings, "lnbits_webpush_privkey", "")
mocker.patch.object(settings, "lnbits_admin_ui", False)
mocker.patch(
"lnbits.core.services.settings.Vapid",
return_value=FakeVapid(has_public_key=False),
)
with pytest.raises(ValueError, match="VAPID public key does not exist"):
await check_webpush_settings()
@pytest.mark.anyio
async def test_check_webpush_settings_skips_generation_when_keys_exist(
settings: Settings, mocker: MockerFixture
):
mocker.patch.object(settings, "lnbits_webpush_privkey", "existing-private-key")
mocker.patch.object(settings, "lnbits_webpush_pubkey", "existing-public-key")
vapid = mocker.patch("lnbits.core.services.settings.Vapid")
update_admin = mocker.patch(
"lnbits.core.services.settings.update_admin_settings",
AsyncMock(),
)
await check_webpush_settings()
vapid.assert_not_called()
update_admin.assert_not_awaited()
tests/unit/test_settings.py
+167 -1
View File
@@ -1,6 +1,18 @@
import pytest
from pytest_mock.plugin import MockerFixture
from lnbits.settings import RedirectPath
from lnbits.settings import (
AssetSettings,
ExchangeRateProvider,
InstalledExtensionsSettings,
NotificationsSettings,
PublicSettings,
RedirectPath,
SecuritySettings,
Settings,
list_parse_fallback,
set_cli_settings,
)
lnurlp_redirect_path = {
"from_path": "/.well-known/lnurlp",
@@ -166,3 +178,157 @@ def test_redirect_path_new_path_from(lnurlp: RedirectPath):
lnurlp.new_path_from("/.well-known/lnurlp/path/more")
== "/lnurlp/api/v1/well-known/path/more"
)
def test_list_parse_fallback():
assert list_parse_fallback("a, b, c") == ["a", "b", "c"]
assert list_parse_fallback('["a", "b"]') == ["a", "b"]
assert list_parse_fallback("") == []
def test_exchange_rate_provider_convert_ticker():
provider = ExchangeRateProvider(
name="Provider",
api_url="https://example.com",
path="$.price",
ticker_conversion=["USD:USDT"],
)
invalid_provider = ExchangeRateProvider(
name="Invalid",
api_url="https://example.com",
path="$.price",
ticker_conversion=["invalid"],
)
assert provider.convert_ticker("USD") == "USDT"
assert provider.convert_ticker("EUR") == "EUR"
assert invalid_provider.convert_ticker("USD") == "USD"
def test_installed_extensions_settings_activate_and_deactivate_paths():
installed = InstalledExtensionsSettings()
redirects = [
{
"from_path": "/.well-known/lnurlp",
"redirect_to_path": "/api/v1/well-known",
}
]
installed.activate_extension_paths(
"lnurlp",
upgrade_hash="hash123",
ext_redirects=redirects,
)
redirect = installed.find_extension_redirect("/.well-known/lnurlp", [])
assert redirect is not None
assert redirect.ext_id == "lnurlp"
assert installed.lnbits_upgraded_extensions["lnurlp"] == "hash123"
assert "lnurlp" in installed.lnbits_installed_extensions_ids
installed.deactivate_extension_paths("lnurlp")
assert "lnurlp" in installed.lnbits_deactivated_extensions
assert installed.find_extension_redirect("/.well-known/lnurlp", []) is None
def test_installed_extensions_settings_detects_conflicting_redirects():
installed = InstalledExtensionsSettings(
lnbits_extensions_redirects=[
RedirectPath(
ext_id="ext_a",
from_path="/.well-known/lnurlp",
redirect_to_path="/api/v1/well-known",
)
]
)
with pytest.raises(ValueError, match="Cannot redirect for extension 'ext_b'"):
installed.activate_extension_paths(
"ext_b",
ext_redirects=[
{
"from_path": "/.well-known/lnurlp",
"redirect_to_path": "/api/v1/well-known",
}
],
)
def test_settings_helper_methods(settings: Settings, mocker: MockerFixture):
mocker.patch.object(settings, "super_user", "super-user")
mocker.patch.object(settings, "lnbits_admin_users", ["admin-user"])
mocker.patch.object(settings, "lnbits_allowed_users", ["allowed-user"])
mocker.patch.object(settings, "lnbits_installed_extensions_ids", {"installed"})
mocker.patch.object(settings, "lnbits_all_extensions_ids", {"installed", "new"})
assert settings.is_user_allowed("allowed-user") is True
assert settings.is_user_allowed("admin-user") is True
assert settings.is_user_allowed("super-user") is True
assert settings.is_user_allowed("random-user") is False
assert settings.is_super_user("super-user") is True
assert settings.is_admin_user("admin-user") is True
assert settings.is_installed_extension_id("installed") is True
assert settings.is_ready_to_install_extension_id("new") is True
assert settings.is_ready_to_install_extension_id("installed") is False
def test_asset_security_and_notification_helpers(
settings: Settings, mocker: MockerFixture
):
mocker.patch.object(settings, "super_user", "super-user")
mocker.patch.object(settings, "lnbits_admin_users", ["admin-user"])
asset_settings = AssetSettings(lnbits_assets_no_limit_users=["vip-user"])
security_settings = SecuritySettings(lnbits_wallet_limit_max_balance=100)
notification_settings = NotificationsSettings(
lnbits_nostr_notifications_enabled=True,
lnbits_nostr_notifications_private_key="nostr-key",
lnbits_telegram_notifications_enabled=True,
lnbits_telegram_notifications_access_token="telegram-token",
)
assert asset_settings.is_unlimited_assets_user("admin-user") is True
assert asset_settings.is_unlimited_assets_user("vip-user") is True
assert asset_settings.is_unlimited_assets_user("random-user") is False
assert security_settings.is_wallet_max_balance_exceeded(101) is True
assert security_settings.is_wallet_max_balance_exceeded(100) is False
assert notification_settings.is_nostr_notifications_configured() is True
assert notification_settings.is_telegram_notifications_configured() is True
def test_public_settings_from_settings(settings: Settings):
original_site_title = settings.lnbits_site_title
original_ad_space = settings.lnbits_ad_space
original_ad_space_enabled = settings.lnbits_ad_space_enabled
original_installed_extensions = settings.lnbits_installed_extensions_ids
original_first_install_token = settings.first_install_token
try:
settings.lnbits_site_title = "Test LNbits"
settings.lnbits_ad_space = "https://example.com;/banner.png;/thumb.png"
settings.lnbits_ad_space_enabled = True
settings.lnbits_installed_extensions_ids = {"ext_a"}
settings.first_install_token = "token"
public = PublicSettings.from_settings(settings)
assert public.site_title == "Test LNbits"
assert public.show_ad_space is True
assert public.ad_space == [["https://example.com", "/banner.png", "/thumb.png"]]
assert set(public.extensions) == {"ext_a"}
assert public.has_first_install_token is True
finally:
settings.lnbits_site_title = original_site_title
settings.lnbits_ad_space = original_ad_space
settings.lnbits_ad_space_enabled = original_ad_space_enabled
settings.lnbits_installed_extensions_ids = original_installed_extensions
settings.first_install_token = original_first_install_token
def test_set_cli_settings_updates_runtime_settings(settings: Settings):
original_host = settings.host
try:
set_cli_settings(host="0.0.0.0")
assert settings.host == "0.0.0.0"
finally:
settings.host = original_host