diff --git a/tests/unit/test_cache.py b/tests/unit/test_cache.py index f7ec4a906..2f33ffe1d 100644 --- a/tests/unit/test_cache.py +++ b/tests/unit/test_cache.py @@ -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") diff --git a/tests/unit/test_crypto_aes.py b/tests/unit/test_crypto_aes.py index 3e098afde..bd0b7ae5d 100644 --- a/tests/unit/test_crypto_aes.py +++ b/tests/unit/test_crypto_aes.py @@ -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) diff --git a/tests/unit/test_decorators.py b/tests/unit/test_decorators.py index 5aa425e58..304aebabf 100644 --- a/tests/unit/test_decorators.py +++ b/tests/unit/test_decorators.py @@ -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) diff --git a/tests/unit/test_exchange_rates.py b/tests/unit/test_exchange_rates.py index 9b3263b49..d90788c2b 100644 --- a/tests/unit/test_exchange_rates.py +++ b/tests/unit/test_exchange_rates.py @@ -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") diff --git a/tests/unit/test_fiat_providers.py b/tests/unit/test_fiat_providers.py index 26ef66661..8ba5f7401 100644 --- a/tests/unit/test_fiat_providers.py +++ b/tests/unit/test_fiat_providers.py @@ -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." diff --git a/tests/unit/test_helpers.py b/tests/unit/test_helpers.py index 924640f80..b98ba761c 100644 --- a/tests/unit/test_helpers.py +++ b/tests/unit/test_helpers.py @@ -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() diff --git a/tests/unit/test_helpers_query.py b/tests/unit/test_helpers_query.py index 5e2469f4a..74bb737d7 100644 --- a/tests/unit/test_helpers_query.py +++ b/tests/unit/test_helpers_query.py @@ -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 diff --git a/tests/unit/test_services_settings.py b/tests/unit/test_services_settings.py new file mode 100644 index 000000000..6a2fe8f25 --- /dev/null +++ b/tests/unit/test_services_settings.py @@ -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() diff --git a/tests/unit/test_settings.py b/tests/unit/test_settings.py index a300afb0f..2855958b1 100644 --- a/tests/unit/test_settings.py +++ b/tests/unit/test_settings.py @@ -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