diff --git a/tests/conftest.py b/tests/conftest.py index 9dd3d9072..6db84adcc 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -45,6 +45,7 @@ def settings(): # override settings for tests lnbits_settings.auth_https_only = False lnbits_settings.lnbits_admin_extensions = [] + lnbits_settings.lnbits_backend_wallet_class = "FakeWallet" lnbits_settings.lnbits_data_folder = "./tests/data" lnbits_settings.lnbits_admin_ui = True lnbits_settings.lnbits_extensions_default_install = [] diff --git a/tests/helpers.py b/tests/helpers.py index 45e7b2b0d..d2ff0471f 100644 --- a/tests/helpers.py +++ b/tests/helpers.py @@ -3,6 +3,7 @@ import string from pydantic import BaseModel +from lnbits.settings import settings from lnbits.wallets import get_funding_source, set_funding_source @@ -39,8 +40,8 @@ def get_random_string(iterations: int = 10): async def get_random_invoice_data(): return {"out": False, "amount": 10, "memo": f"test_memo_{get_random_string(10)}"} - -set_funding_source() +settings.lnbits_backend_wallet_class = "FakeWallet" +set_funding_source("FakeWallet") funding_source = get_funding_source() is_fake: bool = funding_source.__class__.__name__ == "FakeWallet" is_regtest: bool = not is_fake diff --git a/tests/unit/test_services_assets.py b/tests/unit/test_services_assets.py new file mode 100644 index 000000000..0cf2d57da --- /dev/null +++ b/tests/unit/test_services_assets.py @@ -0,0 +1,117 @@ +from io import BytesIO +from types import SimpleNamespace +from uuid import uuid4 + +import pytest +from PIL import Image +from pytest_mock.plugin import MockerFixture + +from lnbits.core.crud import create_account +from lnbits.core.crud.assets import get_user_asset, get_user_assets_count +from lnbits.core.models import Account +from lnbits.core.services.assets import create_user_asset, thumbnail_from_bytes +from lnbits.settings import Settings + + +async def _create_user() -> str: + user_id = uuid4().hex + await create_account(Account(id=user_id, username=f"user_{user_id[:8]}")) + return user_id + + +@pytest.mark.anyio +async def test_create_user_asset_validates_upload_constraints( + settings: Settings, mocker: MockerFixture +): + file_without_type = SimpleNamespace( + content_type=None, + filename="a.txt", + read=mocker.AsyncMock(return_value=b"hello"), + ) + with pytest.raises(ValueError, match="File must have a content type."): + await create_user_asset("user-1", file_without_type, is_public=False) + + bad_type = SimpleNamespace( + content_type="application/x-msdownload", + filename="bad.bin", + read=mocker.AsyncMock(return_value=b"hello"), + ) + with pytest.raises(ValueError, match="File type 'application/x-msdownload' not allowed."): + await create_user_asset("user-1", bad_type, is_public=False) + + original_max_assets = settings.lnbits_max_assets_per_user + original_max_size = settings.lnbits_max_asset_size_mb + original_no_limit_users = list(settings.lnbits_assets_no_limit_users) + try: + settings.lnbits_max_assets_per_user = 1 + settings.lnbits_max_asset_size_mb = 1 + settings.lnbits_assets_no_limit_users = [] + limited_user = await _create_user() + allowed_type = SimpleNamespace( + content_type="text/plain", + filename="ok.txt", + read=mocker.AsyncMock(return_value=b"hello"), + ) + await create_user_asset(limited_user, allowed_type, is_public=False) + + blocked_by_count = SimpleNamespace( + content_type="text/plain", + filename="again.txt", + read=mocker.AsyncMock(return_value=b"again"), + ) + with pytest.raises(ValueError, match="Max upload count of 1 exceeded."): + await create_user_asset(limited_user, blocked_by_count, is_public=False) + + settings.lnbits_max_asset_size_mb = 0.000001 + oversized_user = await _create_user() + large_file = SimpleNamespace( + content_type="text/plain", + filename="ok.txt", + read=mocker.AsyncMock(return_value=b"0123456789"), + ) + with pytest.raises(ValueError, match="File limit of 1e-06MB exceeded."): + await create_user_asset(oversized_user, large_file, is_public=False) + finally: + settings.lnbits_max_assets_per_user = original_max_assets + settings.lnbits_max_asset_size_mb = original_max_size + settings.lnbits_assets_no_limit_users = original_no_limit_users + + +@pytest.mark.anyio +async def test_create_user_asset_success(mocker: MockerFixture): + user_id = await _create_user() + mocker.patch( + "lnbits.core.services.assets.thumbnail_from_bytes", + return_value=None, + ) + file = SimpleNamespace( + content_type="text/plain", + filename="hello.txt", + read=mocker.AsyncMock(return_value=b"hello"), + ) + + asset = await create_user_asset(user_id, file, is_public=True) + stored = await get_user_asset(user_id, asset.id) + + assert asset.id + assert asset.user_id == user_id + assert asset.name == "hello.txt" + assert asset.size_bytes == 5 + assert asset.data == b"hello" + assert asset.is_public is True + assert stored is not None + assert stored.id == asset.id + assert stored.data == b"hello" + assert await get_user_assets_count(user_id) == 1 + + +def test_thumbnail_from_bytes_success_and_failure(): + image = Image.new("RGB", (512, 512), color="red") + buffer = BytesIO() + image.save(buffer, format="PNG") + + thumbnail = thumbnail_from_bytes(buffer.getvalue()) + + assert thumbnail is not None + assert isinstance(thumbnail.getvalue(), bytes) + assert thumbnail_from_bytes(b"not-an-image") is None diff --git a/tests/unit/test_services_extensions.py b/tests/unit/test_services_extensions.py new file mode 100644 index 000000000..b929b9d18 --- /dev/null +++ b/tests/unit/test_services_extensions.py @@ -0,0 +1,307 @@ +from uuid import uuid4 +from types import SimpleNamespace + +import pytest +from pytest_mock.plugin import MockerFixture + +from lnbits.core.crud import ( + create_installed_extension, + delete_installed_extension, + get_installed_extension, + get_installed_extensions, +) +from lnbits.core.models.extensions import ( + Extension, + ExtensionMeta, + ExtensionRelease, + InstallableExtension, + ReleasePaymentInfo, +) +from lnbits.core.services.extensions import ( + activate_extension, + deactivate_extension, + get_valid_extension, + get_valid_extensions, + install_extension, + start_extension_background_work, + stop_extension_background_work, + uninstall_extension, +) +from lnbits.settings import Settings + + +def _installable_extension( + ext_id: str, + version: str = "1.0.0", + compatible: bool = True, + *, + payments: list[ReleasePaymentInfo] | None = None, +) -> InstallableExtension: + return InstallableExtension( + id=ext_id, + name=f"Extension {ext_id}", + version=version, + short_description="Demo extension", + meta=ExtensionMeta( + installed_release=ExtensionRelease( + name=ext_id, + version=version, + archive=f"https://example.com/{ext_id}.zip", + source_repo="org/repo", + hash=f"hash-{ext_id}", + is_version_compatible=compatible, + ), + payments=payments or [], + ), + ) + + +@pytest.mark.anyio +async def test_install_extension_rejects_incompatible_release(tmp_path, settings: Settings): + ext_info = _installable_extension(f"ext_{uuid4().hex[:8]}", compatible=False) + original_data_folder = settings.lnbits_data_folder + original_extensions_path = settings.lnbits_extensions_path + try: + settings.lnbits_data_folder = str(tmp_path / "data") + settings.lnbits_extensions_path = str(tmp_path / "code") + + with pytest.raises(ValueError, match="Incompatible extension version"): + await install_extension(ext_info) + finally: + settings.lnbits_data_folder = original_data_folder + settings.lnbits_extensions_path = original_extensions_path + + +@pytest.mark.anyio +async def test_install_extension_creates_new_extension_and_starts_background_work( + tmp_path, settings: Settings, mocker: MockerFixture +): + ext_id = f"ext_{uuid4().hex[:8]}" + ext_info = _installable_extension(ext_id) + original_data_folder = settings.lnbits_data_folder + original_extensions_path = settings.lnbits_extensions_path + download_mock = mocker.patch.object( + InstallableExtension, "download_archive", mocker.AsyncMock() + ) + extract_mock = mocker.patch.object(InstallableExtension, "extract_archive") + start_mock = mocker.patch( + "lnbits.core.services.extensions.start_extension_background_work", + mocker.AsyncMock(return_value=True), + ) + mocker.patch( + "lnbits.core.services.extensions.get_db_version", + mocker.AsyncMock(return_value=0), + ) + mocker.patch( + "lnbits.core.services.extensions.migrate_extension_database", + mocker.AsyncMock(), + ) + + try: + settings.lnbits_data_folder = str(tmp_path / "data") + settings.lnbits_extensions_path = str(tmp_path / "code") + + extension = await install_extension(ext_info) + stored = await get_installed_extension(ext_id) + finally: + await delete_installed_extension(ext_id=ext_id) + settings.lnbits_data_folder = original_data_folder + settings.lnbits_extensions_path = original_extensions_path + + assert extension.code == ext_id + assert stored is not None + download_mock.assert_awaited_once() + extract_mock.assert_called_once() + start_mock.assert_awaited_once_with(ext_id) + + +@pytest.mark.anyio +async def test_install_extension_updates_existing_upgrade_and_preserves_payments( + tmp_path, settings: Settings, mocker: MockerFixture +): + ext_id = f"ext_{uuid4().hex[:8]}" + existing_payment = ReleasePaymentInfo( + pay_link="https://pay.example", + payment_hash="payment-hash", + ) + existing_ext = _installable_extension(ext_id, payments=[existing_payment]) + updated_ext = _installable_extension(ext_id, version="2.0.0") + original_data_folder = settings.lnbits_data_folder + original_extensions_path = settings.lnbits_extensions_path + extract_mock = mocker.patch.object(InstallableExtension, "extract_archive") + start_mock = mocker.patch( + "lnbits.core.services.extensions.start_extension_background_work", + mocker.AsyncMock(return_value=True), + ) + stop_mock = mocker.patch( + "lnbits.core.services.extensions.stop_extension_background_work", + mocker.AsyncMock(return_value=True), + ) + mocker.patch( + "lnbits.core.services.extensions.get_db_version", + mocker.AsyncMock(return_value=1), + ) + mocker.patch( + "lnbits.core.services.extensions.migrate_extension_database", + mocker.AsyncMock(), + ) + + try: + settings.lnbits_data_folder = str(tmp_path / "data") + settings.lnbits_extensions_path = str(tmp_path / "code") + await create_installed_extension(existing_ext) + updated_ext.ext_upgrade_dir.mkdir(parents=True, exist_ok=True) + + extension = await install_extension(updated_ext, skip_download=True) + stored = await get_installed_extension(ext_id) + finally: + await delete_installed_extension(ext_id=ext_id) + settings.lnbits_data_folder = original_data_folder + settings.lnbits_extensions_path = original_extensions_path + + assert extension.code == ext_id + assert extension.is_upgrade_extension is True + assert stored is not None + assert stored.meta is not None + assert stored.meta.payments == [existing_payment] + extract_mock.assert_called_once() + stop_mock.assert_awaited_once_with(ext_id) + start_mock.assert_awaited_once_with(ext_id) + + +@pytest.mark.anyio +async def test_uninstall_activate_and_deactivate_extensions( + tmp_path, settings: Settings, mocker: MockerFixture +): + ext_id = f"ext_{uuid4().hex[:8]}" + ext_info = _installable_extension(ext_id) + original_data_folder = settings.lnbits_data_folder + original_extensions_path = settings.lnbits_extensions_path + original_deactivated = set(settings.lnbits_deactivated_extensions) + stop_mock = mocker.patch( + "lnbits.core.services.extensions.stop_extension_background_work", + mocker.AsyncMock(return_value=True), + ) + start_mock = mocker.patch( + "lnbits.core.services.extensions.start_extension_background_work", + mocker.AsyncMock(return_value=True), + ) + clean_mock = mocker.patch.object(InstallableExtension, "clean_extension_files") + register_routes_mock = mocker.patch( + "lnbits.core.services.extensions.core_app_extra.register_new_ext_routes" + ) + + try: + settings.lnbits_data_folder = str(tmp_path / "data") + settings.lnbits_extensions_path = str(tmp_path / "code") + await create_installed_extension(ext_info) + + await uninstall_extension(ext_id) + assert await get_installed_extension(ext_id) is None + assert ext_id in settings.lnbits_deactivated_extensions + + await create_installed_extension(ext_info) + await activate_extension(Extension(code=ext_id, is_valid=True)) + active_ext = await get_installed_extension(ext_id) + assert active_ext is not None + assert active_ext.active is True + + await deactivate_extension(ext_id) + inactive_ext = await get_installed_extension(ext_id) + assert inactive_ext is not None + assert inactive_ext.active is False + assert ext_id in settings.lnbits_deactivated_extensions + finally: + await delete_installed_extension(ext_id=ext_id) + settings.lnbits_data_folder = original_data_folder + settings.lnbits_extensions_path = original_extensions_path + settings.lnbits_deactivated_extensions = original_deactivated + + clean_mock.assert_called_once() + register_routes_mock.assert_called_once() + assert stop_mock.await_count == 2 + assert start_mock.await_count == 1 + + +@pytest.mark.anyio +async def test_stop_extension_background_work_handles_missing_and_async_stops( + mocker: MockerFixture, +): + import_module_mock = mocker.patch( + "lnbits.core.services.extensions.importlib.import_module", + return_value=object(), + ) + + assert await stop_extension_background_work("demoext") is False + + called = {"stop": False} + + async def demoext_stop(): + called["stop"] = True + + import_module_mock.return_value = SimpleNamespace(demoext_stop=demoext_stop) + + assert await stop_extension_background_work("demoext") is True + assert called["stop"] is True + + +@pytest.mark.anyio +async def test_start_extension_background_work_handles_missing_and_sync_starts( + mocker: MockerFixture, +): + import_module_mock = mocker.patch( + "lnbits.core.services.extensions.importlib.import_module", + return_value=object(), + ) + + assert await start_extension_background_work("demoext") is False + + called = {"start": False} + + def demoext_start(): + called["start"] = True + + import_module_mock.return_value = SimpleNamespace(demoext_start=demoext_start) + + assert await start_extension_background_work("demoext") is True + assert called["start"] is True + + +@pytest.mark.anyio +async def test_get_valid_extensions_and_single_extension_respect_settings( + tmp_path, settings: Settings +): + ext_id_one = f"ext_{uuid4().hex[:8]}" + ext_id_two = f"ext_{uuid4().hex[:8]}" + ext_one = _installable_extension(ext_id_one) + ext_two = _installable_extension(ext_id_two) + original_deactivated = set(settings.lnbits_deactivated_extensions) + original_deactivate_all = settings.lnbits_extensions_deactivate_all + original_data_folder = settings.lnbits_data_folder + original_extensions_path = settings.lnbits_extensions_path + + try: + settings.lnbits_data_folder = str(tmp_path / "data") + settings.lnbits_extensions_path = str(tmp_path / "code") + settings.lnbits_deactivated_extensions = {ext_id_two} + settings.lnbits_extensions_deactivate_all = False + await create_installed_extension(ext_one) + await create_installed_extension(ext_two) + + valid_extensions = await get_valid_extensions(include_deactivated=False) + valid_codes = {ext.code for ext in valid_extensions} + assert ext_id_one in valid_codes + assert ext_id_two not in valid_codes + + assert await get_valid_extension(ext_id_one, include_deactivated=True) is not None + + settings.lnbits_extensions_deactivate_all = True + assert await get_valid_extensions(include_deactivated=False) == [] + assert await get_valid_extension(ext_id_one, include_deactivated=False) is None + finally: + await delete_installed_extension(ext_id=ext_id_one) + await delete_installed_extension(ext_id=ext_id_two) + settings.lnbits_deactivated_extensions = original_deactivated + settings.lnbits_extensions_deactivate_all = original_deactivate_all + settings.lnbits_data_folder = original_data_folder + settings.lnbits_extensions_path = original_extensions_path diff --git a/tests/unit/test_services_extensions_builder.py b/tests/unit/test_services_extensions_builder.py new file mode 100644 index 000000000..ece1c5547 --- /dev/null +++ b/tests/unit/test_services_extensions_builder.py @@ -0,0 +1,139 @@ +import hashlib +import zipfile +from pathlib import Path +from types import SimpleNamespace + +import pytest +from pytest_mock.plugin import MockerFixture + +from lnbits.core.models.extensions import ExtensionRelease +from lnbits.core.models.extensions_builder import ( + ActionFields, + ClientDataFields, + DataField, + DataFields, + ExtensionData, + OwnerDataFields, + PublicPageFields, + SettingsFields, +) +from lnbits.core.services.extensions_builder import ( + build_extension_from_data, + clean_extension_builder_data, + zip_directory, +) +from lnbits.settings import Settings + + +def _extension_data() -> ExtensionData: + return ExtensionData( + id="demoext", + name="Demo Extension", + stub_version="0.1.0", + short_description="Generated extension", + owner_data=DataFields( + name="OwnerData", + fields=[DataField(name="wallet_id", type="wallet")], + ), + client_data=DataFields( + name="ClientData", + fields=[DataField(name="amount", type="int")], + ), + settings_data=SettingsFields(name="SettingsData", fields=[]), + public_page=PublicPageFields( + owner_data_fields=OwnerDataFields(), + client_data_fields=ClientDataFields(), + action_fields=ActionFields(), + ), + ) + + +@pytest.mark.anyio +async def test_build_extension_from_data_orchestrates_builder_steps( + tmp_path, mocker: MockerFixture +): + data = _extension_data() + release = ExtensionRelease( + name="stub", + version="0.1.0", + archive="https://example.com/stub.zip", + source_repo="org/repo", + is_github_release=True, + ) + build_dir = tmp_path / "build" + fetch_mock = mocker.patch( + "lnbits.core.services.extensions_builder._fetch_extension_builder_stub", + mocker.AsyncMock(), + ) + transform_mock = mocker.patch( + "lnbits.core.services.extensions_builder._transform_extension_builder_stub" + ) + export_mock = mocker.patch( + "lnbits.core.services.extensions_builder._export_extension_data_json" + ) + mocker.patch( + "lnbits.core.services.extensions_builder._get_extension_stub_release", + mocker.AsyncMock(return_value=release), + ) + mocker.patch( + "lnbits.core.services.extensions_builder._copy_ext_stub_to_build_dir", + return_value=build_dir, + ) + mocker.patch( + "lnbits.core.services.extensions_builder.uuid4", + return_value=SimpleNamespace(hex="seed"), + ) + + built_release, output_dir = await build_extension_from_data(data, "stub-ext") + + assert output_dir == build_dir + assert built_release == release + assert built_release.hash == hashlib.sha256(b"seed").hexdigest() + assert built_release.icon == "/demoext/static/image/demoext.png" + assert built_release.is_github_release is False + fetch_mock.assert_awaited_once_with("stub-ext", release) + transform_mock.assert_called_once_with(data, build_dir) + export_mock.assert_called_once_with(data, build_dir) + + +def test_clean_extension_builder_data_recreates_working_directory( + settings: Settings, tmp_path +): + original_data_folder = settings.lnbits_data_folder + try: + settings.lnbits_data_folder = str(tmp_path) + working_dir = settings.extension_builder_working_dir_path + working_dir.mkdir(parents=True, exist_ok=True) + Path(working_dir, "stale.txt").write_text("stale") + + clean_extension_builder_data() + + assert working_dir.is_dir() + assert list(working_dir.iterdir()) == [] + finally: + settings.lnbits_data_folder = original_data_folder + + +def test_zip_directory_skips_excluded_directories(tmp_path): + source_dir = tmp_path / "source" + zip_path = tmp_path / "archive.zip" + (source_dir / "nested").mkdir(parents=True) + (source_dir / "node_modules").mkdir() + (source_dir / "__pycache__").mkdir() + (source_dir / "root.txt").write_text("root") + (source_dir / "nested" / "file.txt").write_text("nested") + (source_dir / "node_modules" / "ignored.txt").write_text("ignored") + (source_dir / "__pycache__" / "ignored.pyc").write_text("ignored") + + from_builder = "lnbits.core.services.extensions_builder._is_excluded_dir" + with pytest.MonkeyPatch.context() as monkeypatch: + monkeypatch.setattr( + from_builder, + lambda path: "node_modules" in path or "__pycache__" in path, + ) + zip_directory(source_dir, zip_path) + + with zipfile.ZipFile(zip_path) as archive: + names = sorted(archive.namelist()) + + assert names == ["nested/file.txt", "root.txt"] diff --git a/tests/unit/test_services_funding_source.py b/tests/unit/test_services_funding_source.py new file mode 100644 index 000000000..1c36b6e4d --- /dev/null +++ b/tests/unit/test_services_funding_source.py @@ -0,0 +1,169 @@ +from types import SimpleNamespace +from uuid import uuid4 + +import pytest +from pytest_mock.plugin import MockerFixture + +from lnbits.core.crud import create_account, create_wallet, get_total_balance +from lnbits.core.models import Account +from lnbits.core.models.misc import BalanceDelta +from lnbits.core.services.funding_source import ( + check_balance_delta_changed, + check_server_balance_against_node, + get_balance_delta, + switch_to_voidwallet, +) +from lnbits.core.services.payments import update_wallet_balance +from lnbits.settings import Settings + + +async def _create_wallet_with_balance(amount: int): + user_id = uuid4().hex + await create_account(Account(id=user_id, username=f"user_{user_id[:8]}")) + wallet = await create_wallet(user_id=user_id, wallet_name="wallet") + await update_wallet_balance(wallet=wallet, amount=amount) + return wallet + + +@pytest.mark.anyio +async def test_switch_to_voidwallet_returns_when_already_using_voidwallet( + settings: Settings, mocker: MockerFixture +): + original_backend_class = settings.lnbits_backend_wallet_class + try: + settings.lnbits_backend_wallet_class = "FakeWallet" + mocker.patch( + "lnbits.core.services.funding_source.get_funding_source", + return_value=type("VoidWallet", (), {})(), + ) + set_funding_source = mocker.patch( + "lnbits.core.services.funding_source.set_funding_source" + ) + + await switch_to_voidwallet() + + set_funding_source.assert_not_called() + assert settings.lnbits_backend_wallet_class == "FakeWallet" + finally: + settings.lnbits_backend_wallet_class = original_backend_class + + +@pytest.mark.anyio +async def test_switch_to_voidwallet_updates_backend_class( + settings: Settings, mocker: MockerFixture +): + original_backend_class = settings.lnbits_backend_wallet_class + try: + settings.lnbits_backend_wallet_class = "FakeWallet" + mocker.patch( + "lnbits.core.services.funding_source.get_funding_source", + return_value=type("FakeWallet", (), {})(), + ) + set_funding_source = mocker.patch( + "lnbits.core.services.funding_source.set_funding_source" + ) + + await switch_to_voidwallet() + + set_funding_source.assert_called_once_with("VoidWallet") + assert settings.lnbits_backend_wallet_class == "VoidWallet" + finally: + settings.lnbits_backend_wallet_class = original_backend_class + + +@pytest.mark.anyio +async def test_get_balance_delta(mocker: MockerFixture): + baseline_balance = await get_total_balance() + await _create_wallet_with_balance(11) + funding_source = SimpleNamespace( + status=mocker.AsyncMock( + return_value=SimpleNamespace(balance_msat=7_000, error_message=None) + ) + ) + mocker.patch( + "lnbits.core.services.funding_source.get_funding_source", + return_value=funding_source, + ) + + delta = await get_balance_delta() + expected_balance_sats = (baseline_balance + 11_000) // 1000 + + assert delta.lnbits_balance_sats == expected_balance_sats + assert delta.node_balance_sats == 7 + assert delta.delta_sats == expected_balance_sats - 7 + + +@pytest.mark.anyio +async def test_check_server_balance_against_node_notifies_and_switches( + settings: Settings, mocker: MockerFixture +): + original_switch = settings.lnbits_watchdog_switch_to_voidwallet + original_notification = settings.lnbits_notification_watchdog + original_delta = settings.lnbits_watchdog_delta + try: + settings.lnbits_watchdog_switch_to_voidwallet = True + settings.lnbits_notification_watchdog = True + settings.lnbits_watchdog_delta = 5 + mocker.patch( + "lnbits.core.services.funding_source.get_funding_source", + return_value=type("FakeWallet", (), {})(), + ) + mocker.patch( + "lnbits.core.services.funding_source.get_balance_delta", + mocker.AsyncMock( + return_value=BalanceDelta( + lnbits_balance_sats=12, + node_balance_sats=1, + ) + ), + ) + enqueue = mocker.patch( + "lnbits.core.services.funding_source.enqueue_admin_notification" + ) + switch = mocker.patch( + "lnbits.core.services.funding_source.switch_to_voidwallet", + mocker.AsyncMock(), + ) + + await check_server_balance_against_node() + + enqueue.assert_called_once() + switch.assert_awaited_once() + finally: + settings.lnbits_watchdog_switch_to_voidwallet = original_switch + settings.lnbits_notification_watchdog = original_notification + settings.lnbits_watchdog_delta = original_delta + + +@pytest.mark.anyio +async def test_check_balance_delta_changed_tracks_and_notifies( + settings: Settings, mocker: MockerFixture +): + original_latest = settings.latest_balance_delta_sats + original_threshold = settings.notification_balance_delta_threshold_sats + try: + settings.latest_balance_delta_sats = None + settings.notification_balance_delta_threshold_sats = 3 + mocker.patch( + "lnbits.core.services.funding_source.get_balance_delta", + mocker.AsyncMock( + side_effect=[ + BalanceDelta(lnbits_balance_sats=12, node_balance_sats=10), + BalanceDelta(lnbits_balance_sats=20, node_balance_sats=10), + ] + ), + ) + enqueue = mocker.patch( + "lnbits.core.services.funding_source.enqueue_admin_notification" + ) + + await check_balance_delta_changed() + enqueue.assert_not_called() + assert settings.latest_balance_delta_sats == 2 + + await check_balance_delta_changed() + enqueue.assert_called_once() + assert settings.latest_balance_delta_sats == 10 + finally: + settings.latest_balance_delta_sats = original_latest + settings.notification_balance_delta_threshold_sats = original_threshold diff --git a/tests/unit/test_services_lnurl.py b/tests/unit/test_services_lnurl.py new file mode 100644 index 000000000..42aac8e38 --- /dev/null +++ b/tests/unit/test_services_lnurl.py @@ -0,0 +1,175 @@ +from uuid import uuid4 + +import pytest +from lnurl import ( + LnAddress, + LnurlErrorResponse, + LnurlPayActionResponse, + LnurlPayResponse, + LnurlResponseException, + LnurlSuccessResponse, + LnurlWithdrawResponse, +) +from pytest_mock.plugin import MockerFixture + +from lnbits.core.crud import create_account, create_wallet, get_wallet +from lnbits.core.models import Account +from lnbits.core.models.lnurl import CreateLnurlPayment +from lnbits.core.models.wallets import Wallet +from lnbits.core.services.lnurl import ( + fetch_lnurl_pay_request, + get_pr_from_lnurl, + perform_withdraw, + store_paylink, +) + +TEST_BOLT11 = ( + "lnbc1pnsu5z3pp57getmdaxhg5kc9yh2a2qsh7cjf4gnccgkw0qenm8vsqv50w7s" + "ygqdqj0fjhymeqv9kk7atwwscqzzsxqyz5vqsp5e2yyqcp0a3ujeesp24ya0glej" + "srh703md8mrx0g2lyvjxy5w27ss9qxpqysgqyjreasng8a086kpkczv48er5c6l5" + "73aym6ynrdl9nkzqnag49vt3sjjn8qdfq5cr6ha0vrdz5c5r3v4aghndly0hplmv" + "6hjxepwp93cq398l3s" +) + + +def _make_pay_response() -> LnurlPayResponse: + return LnurlPayResponse( + callback="https://example.com/callback", + minSendable=1, + maxSendable=10_000, + metadata='[["text/plain","Test"],["text/identifier","alice@example.com"]]', + ) + + +def _make_wallet() -> Wallet: + return Wallet( + id="wallet-id", + user="user-id", + name="Wallet", + adminkey="admin-key", + inkey="invoice-key", + ) + + +async def _create_wallet() -> Wallet: + user_id = uuid4().hex + await create_account(Account(id=user_id, username=f"user_{user_id[:8]}")) + return await create_wallet(user_id=user_id, wallet_name="Wallet") + + +@pytest.mark.anyio +async def test_perform_withdraw_success_and_validation(mocker: MockerFixture): + withdraw_response = LnurlWithdrawResponse( + callback="https://example.com/callback", + k1="k1", + minWithdrawable=1, + maxWithdrawable=1000, + defaultDescription="test", + ) + execute_withdraw_mock = mocker.patch( + "lnbits.core.services.lnurl.execute_withdraw", + mocker.AsyncMock(return_value=LnurlSuccessResponse()), + ) + mocker.patch( + "lnbits.core.services.lnurl.handle", + mocker.AsyncMock(return_value=withdraw_response), + ) + + await perform_withdraw("lnurl", "bolt11") + + execute_withdraw_mock.assert_awaited_once() + + mocker.patch( + "lnbits.core.services.lnurl.check_callback_url", + side_effect=ValueError("blocked"), + ) + with pytest.raises(LnurlResponseException, match="Invalid callback URL"): + await perform_withdraw("lnurl", "bolt11") + + +@pytest.mark.anyio +async def test_perform_withdraw_rejects_error_response(mocker: MockerFixture): + mocker.patch( + "lnbits.core.services.lnurl.handle", + mocker.AsyncMock(return_value=LnurlErrorResponse(reason="boom")), + ) + + with pytest.raises(LnurlResponseException, match="boom"): + await perform_withdraw("lnurl", "bolt11") + + +@pytest.mark.anyio +async def test_get_pr_from_lnurl_success_and_error(mocker: MockerFixture): + pay_response = _make_pay_response() + mocker.patch( + "lnbits.core.services.lnurl.handle", + mocker.AsyncMock(return_value=pay_response), + ) + mocker.patch( + "lnbits.core.services.lnurl.execute_pay_request", + mocker.AsyncMock(return_value=LnurlPayActionResponse(pr=TEST_BOLT11)), + ) + + assert await get_pr_from_lnurl("lnurl", 1000, comment="hello") == TEST_BOLT11 + + mocker.patch( + "lnbits.core.services.lnurl.handle", + mocker.AsyncMock(return_value=LnurlErrorResponse(reason="nope")), + ) + with pytest.raises(LnurlResponseException, match="nope"): + await get_pr_from_lnurl("lnurl", 1000) + + +@pytest.mark.anyio +async def test_fetch_lnurl_pay_request_converts_currency_and_stores_paylink( + mocker: MockerFixture, +): + pay_response = _make_pay_response() + action_response = LnurlPayActionResponse(pr=TEST_BOLT11, disposable=False) + mocker.patch( + "lnbits.core.services.lnurl.fiat_amount_as_satoshis", + mocker.AsyncMock(return_value=100), + ) + execute_mock = mocker.patch( + "lnbits.core.services.lnurl.execute_pay_request", + mocker.AsyncMock(return_value=action_response), + ) + store_paylink_mock = mocker.patch( + "lnbits.core.services.lnurl.store_paylink", + mocker.AsyncMock(), + ) + wallet = _make_wallet() + + data = CreateLnurlPayment(res=pay_response, amount=2500, unit="USD", comment="hi") + response, action = await fetch_lnurl_pay_request(data, wallet=wallet) + + assert response == pay_response + assert action == action_response + execute_mock.assert_awaited_once() + assert execute_mock.await_args.kwargs["msat"] == 100_000 + store_paylink_mock.assert_awaited_once_with(pay_response, action_response, wallet, None) + + with pytest.raises(LnurlResponseException, match="No LNURL pay request provided."): + await fetch_lnurl_pay_request(CreateLnurlPayment(amount=1)) + + +@pytest.mark.anyio +async def test_store_paylink_appends_and_updates_existing(): + wallet = await _create_wallet() + pay_response = _make_pay_response() + action_response = LnurlPayActionResponse(pr=TEST_BOLT11, disposable=False) + + await store_paylink(pay_response, action_response, wallet, LnAddress("alice@example.com")) + stored_wallet = await get_wallet(wallet.id) + + assert stored_wallet is not None + assert len(stored_wallet.stored_paylinks.links) == 1 + assert stored_wallet.stored_paylinks.links[0].lnurl == "alice@example.com" + + first_used = stored_wallet.stored_paylinks.links[0].last_used + await store_paylink(pay_response, action_response, wallet, LnAddress("alice@example.com")) + stored_wallet = await get_wallet(wallet.id) + + assert stored_wallet is not None + assert len(stored_wallet.stored_paylinks.links) == 1 + assert stored_wallet.stored_paylinks.links[0].last_used >= first_used diff --git a/tests/unit/test_services_nostr.py b/tests/unit/test_services_nostr.py new file mode 100644 index 000000000..c3b7c64d5 --- /dev/null +++ b/tests/unit/test_services_nostr.py @@ -0,0 +1,120 @@ +from types import SimpleNamespace + +import httpx +import pytest +from pytest_mock.plugin import MockerFixture + +from lnbits.core.services.nostr import fetch_nip5_details, send_nostr_dm + + +class FakeWebSocket: + def __init__(self): + self.sent: list[str] = [] + self.closed = False + + def send(self, message: str): + self.sent.append(message) + + def close(self): + self.closed = True + + +class MockHTTPClient: + def __init__(self, response): + self.response = response + self.calls: list[str] = [] + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, tb): + return False + + async def get(self, url: str): + self.calls.append(url) + return self.response + + +class MockHTTPResponse: + def __init__(self, json_data: dict, error: Exception | None = None): + self._json_data = json_data + self._error = error + + def raise_for_status(self): + if self._error: + raise self._error + + def json(self): + return self._json_data + + +@pytest.mark.anyio +async def test_send_nostr_dm_sends_to_available_relays_and_closes_connections( + mocker: MockerFixture, +): + event = mocker.Mock() + event.to_message.return_value = "nostr-message" + event.to_dict.return_value = {"id": "event-id"} + dm = mocker.Mock() + dm.to_event.return_value = event + mocker.patch("lnbits.core.services.nostr.EncryptedDirectMessage", return_value=dm) + + ws_one = FakeWebSocket() + ws_two = FakeWebSocket() + mocker.patch( + "lnbits.core.services.nostr.create_connection", + side_effect=[ws_one, RuntimeError("boom"), ws_two], + ) + mocker.patch("lnbits.core.services.nostr.asyncio.sleep", mocker.AsyncMock()) + + result = await send_nostr_dm( + "privkey", + "pubkey", + "hello", + ["wss://relay-1", "wss://broken", "wss://relay-2"], + ) + + assert ws_one.sent == ["nostr-message"] + assert ws_two.sent == ["nostr-message"] + assert ws_one.closed is True + assert ws_two.closed is True + assert result == {"id": "event-id"} + + +@pytest.mark.anyio +async def test_fetch_nip5_details_returns_pubkey_and_relays(mocker: MockerFixture): + response = MockHTTPResponse( + { + "names": {"alice": "f" * 64}, + "relays": {"f" * 64: ["wss://relay.example.com"]}, + } + ) + client = MockHTTPClient(response) + mocker.patch("lnbits.core.services.nostr.is_valid_url", return_value=True) + validate_identifier = mocker.patch("lnbits.core.services.nostr.validate_identifier") + validate_pub_key = mocker.patch("lnbits.core.services.nostr.validate_pub_key") + mocker.patch("lnbits.core.services.nostr.httpx.AsyncClient", return_value=client) + + pubkey, relays = await fetch_nip5_details("alice@example.com") + + validate_identifier.assert_called_once_with("alice") + validate_pub_key.assert_called_once_with("f" * 64) + assert client.calls == ["https://example.com/.well-known/nostr.json?name=alice"] + assert pubkey == "f" * 64 + assert relays == ["wss://relay.example.com"] + + +@pytest.mark.anyio +async def test_fetch_nip5_details_rejects_invalid_values(mocker: MockerFixture): + with pytest.raises(ValueError, match="not enough values to unpack"): + await fetch_nip5_details("invalid") + + mocker.patch("lnbits.core.services.nostr.is_valid_url", return_value=False) + with pytest.raises(ValueError, match="Invalid NIP5 domain"): + await fetch_nip5_details("alice@example.com") + + mocker.patch("lnbits.core.services.nostr.is_valid_url", return_value=True) + client = MockHTTPClient(MockHTTPResponse({"names": {}})) + mocker.patch("lnbits.core.services.nostr.httpx.AsyncClient", return_value=client) + with pytest.raises(ValueError, match="NIP5 not name found"): + await fetch_nip5_details("alice@example.com") diff --git a/tests/unit/test_services_notifications.py b/tests/unit/test_services_notifications.py new file mode 100644 index 000000000..bb8f6a54c --- /dev/null +++ b/tests/unit/test_services_notifications.py @@ -0,0 +1,598 @@ +import asyncio +from datetime import datetime, timezone +from http import HTTPStatus +from types import SimpleNamespace +from unittest.mock import MagicMock +from uuid import uuid4 + +import httpx +import pytest +from pywebpush import WebPushException +from pytest_mock.plugin import MockerFixture + +from lnbits.core.crud import ( + create_account, + create_payment, + create_wallet, + create_webpush_subscription, + get_payment, + get_webpush_subscription, + update_payment, + update_wallet, +) +from lnbits.core.models import Account, CreatePayment, Payment, PaymentState, Wallet +from lnbits.core.models.notifications import NotificationType +from lnbits.core.models.users import UserExtra, UserNotifications +from lnbits.core.models.wallets import ( + WalletPermission, + WalletSharePermission, + WalletShareStatus, +) +from lnbits.core.services.notifications import ( + dispatch_webhook, + enqueue_admin_notification, + enqueue_user_notification, + process_next_notification, + send_admin_notification, + send_chat_payment_notification, + send_email, + send_email_notification, + send_notification, + send_nostr_notification, + send_nostr_notifications, + send_payment_notification, + send_payment_push_notification, + send_push_notification, + send_telegram_message, + send_telegram_notification, + send_user_notification, + send_ws_payment_notification, +) +from lnbits.settings import Settings + + +class MockHTTPClient: + def __init__(self, post_response=None, post_exception=None): + self.post_response = post_response + self.post_exception = post_exception + self.posts: list[tuple[str, dict]] = [] + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, tb): + return None + + async def post(self, url, **kwargs): + self.posts.append((url, kwargs)) + if self.post_exception: + raise self.post_exception + return self.post_response + + +async def _create_wallet( + notifications: UserNotifications | None = None, + *, + name: str | None = None, +) -> Wallet: + account = Account( + id=uuid4().hex, + username=f"user_{uuid4().hex[:8]}", + extra=UserExtra(notifications=notifications or UserNotifications()), + ) + await create_account(account) + return await create_wallet( + user_id=account.id, + wallet_name=name or f"wallet_{account.id[:8]}", + ) + + +async def _create_payment( + wallet: Wallet, + *, + amount_msat: int = 2_000, + status: PaymentState = PaymentState.SUCCESS, + webhook: str | None = None, + webhook_status: str | None = None, + memo: str | None = "memo", + extra: dict | None = None, +) -> Payment: + checking_id = f"checking_{uuid4().hex[:8]}" + payment = await create_payment( + checking_id=checking_id, + data=CreatePayment( + wallet_id=wallet.id, + payment_hash=uuid4().hex, + bolt11=f"bolt11-{checking_id}", + amount_msat=amount_msat, + memo=memo or "", + webhook=webhook, + extra=extra or {}, + ), + status=status, + ) + if webhook_status is not None: + payment.webhook_status = webhook_status + await update_payment(payment) + return await get_payment(checking_id) + + +@pytest.mark.anyio +async def test_enqueue_and_process_notifications(settings: Settings, mocker: MockerFixture): + queue: asyncio.Queue = asyncio.Queue() + admin_mock = mocker.patch( + "lnbits.core.services.notifications.send_admin_notification", + mocker.AsyncMock(), + ) + user_mock = mocker.patch( + "lnbits.core.services.notifications.send_user_notification", + mocker.AsyncMock(), + ) + mocker.patch("lnbits.core.services.notifications.notifications_queue", queue) + mocker.patch( + "lnbits.core.services.notifications._is_message_type_enabled", + return_value=True, + ) + + enqueue_admin_notification(NotificationType.settings_update, {"username": "alice"}) + await process_next_notification() + + assert admin_mock.await_count == 1 + assert admin_mock.await_args.args[0].startswith(f"[{settings.lnbits_site_title}]") + assert "alice" in admin_mock.await_args.args[0] + assert admin_mock.await_args.args[1] == NotificationType.settings_update.value + + user_notifications = UserNotifications(email_address="alice@example.com") + enqueue_user_notification( + NotificationType.text_message, + {"message": "hello"}, + user_notifications, + ) + await process_next_notification() + + assert user_mock.await_count == 1 + assert user_mock.await_args.args[0] == user_notifications + assert "hello" in user_mock.await_args.args[1] + assert user_mock.await_args.args[2] == NotificationType.text_message.value + + +@pytest.mark.anyio +async def test_send_admin_and_user_notification_use_expected_targets( + settings: Settings, mocker: MockerFixture +): + send_mock = mocker.patch( + "lnbits.core.services.notifications.send_notification", + mocker.AsyncMock(), + ) + original_chat_id = settings.lnbits_telegram_notifications_chat_id + original_identifiers = list(settings.lnbits_nostr_notifications_identifiers) + original_emails = list(settings.lnbits_email_notifications_to_emails) + try: + settings.lnbits_telegram_notifications_chat_id = "chat-id" + settings.lnbits_nostr_notifications_identifiers = ["alice@example.com"] + settings.lnbits_email_notifications_to_emails = ["admin@example.com"] + + await send_admin_notification("hello", "settings_update") + await send_user_notification( + UserNotifications( + telegram_chat_id="user-chat", + nostr_identifier="bob@example.com", + email_address="bob@example.com", + ), + "hello user", + "text_message", + ) + finally: + settings.lnbits_telegram_notifications_chat_id = original_chat_id + settings.lnbits_nostr_notifications_identifiers = original_identifiers + settings.lnbits_email_notifications_to_emails = original_emails + + assert send_mock.await_args_list[0].args == ( + "chat-id", + ["alice@example.com"], + ["admin@example.com"], + "hello", + "settings_update", + ) + assert send_mock.await_args_list[1].args == ( + "user-chat", + ["bob@example.com"], + ["bob@example.com"], + "hello user", + "text_message", + ) + + +@pytest.mark.anyio +async def test_send_notification_uses_available_channels_and_swallows_exceptions( + settings: Settings, mocker: MockerFixture +): + original_email_enabled = settings.lnbits_email_notifications_enabled + try: + settings.lnbits_email_notifications_enabled = True + mocker.patch.object( + type(settings), + "is_telegram_notifications_configured", + return_value=True, + ) + mocker.patch.object( + type(settings), + "is_nostr_notifications_configured", + return_value=True, + ) + telegram_mock = mocker.patch( + "lnbits.core.services.notifications.send_telegram_notification", + mocker.AsyncMock(side_effect=Exception("telegram boom")), + ) + nostr_mock = mocker.patch( + "lnbits.core.services.notifications.send_nostr_notifications", + mocker.AsyncMock(return_value=["alice@example.com"]), + ) + email_mock = mocker.patch( + "lnbits.core.services.notifications.send_email_notification", + mocker.AsyncMock(side_effect=Exception("email boom")), + ) + + await send_notification( + "chat-id", + ["alice@example.com"], + ["alice@example.com"], + "hello", + "text_message", + ) + finally: + settings.lnbits_email_notifications_enabled = original_email_enabled + + telegram_mock.assert_awaited_once() + nostr_mock.assert_awaited_once() + email_mock.assert_awaited_once() + + +@pytest.mark.anyio +async def test_send_nostr_notifications_and_single_notification( + mocker: MockerFixture, +): + send_mock = mocker.patch( + "lnbits.core.services.notifications.send_nostr_notification", + mocker.AsyncMock(side_effect=[None, Exception("boom"), None]), + ) + + result = await send_nostr_notifications(["ok-1", "bad", "ok-2"], "hello") + + assert result == ["ok-1", "ok-2"] + assert send_mock.await_count == 3 + + fetch_mock = mocker.patch( + "lnbits.core.services.notifications.fetch_nip5_details", + mocker.AsyncMock(return_value=("pubkey", ["wss://relay"])), + ) + normalize_mock = mocker.patch( + "lnbits.core.services.notifications.normalize_private_key", + return_value="server-private-key", + ) + dm_mock = mocker.patch( + "lnbits.core.services.notifications.send_nostr_dm", + mocker.AsyncMock(), + ) + + await send_nostr_notification("alice@example.com", "hello") + + fetch_mock.assert_awaited_once_with("alice@example.com") + normalize_mock.assert_called_once() + dm_mock.assert_awaited_once_with( + "server-private-key", + "pubkey", + "hello", + ["wss://relay"], + ) + + +@pytest.mark.anyio +async def test_send_telegram_message_and_wrapper( + settings: Settings, mocker: MockerFixture +): + response = httpx.Response( + 200, + request=httpx.Request("POST", "https://api.telegram.org"), + json={"ok": True}, + ) + client = MockHTTPClient(post_response=response) + mocker.patch( + "lnbits.core.services.notifications.httpx.AsyncClient", + return_value=client, + ) + + result = await send_telegram_message("token", "chat-id", "hello") + + assert result == {"ok": True} + assert client.posts[0][0].endswith("/bottoken/sendMessage") + + original_token = settings.lnbits_telegram_notifications_access_token + try: + settings.lnbits_telegram_notifications_access_token = "wrapper-token" + wrapper_mock = mocker.patch( + "lnbits.core.services.notifications.send_telegram_message", + mocker.AsyncMock(return_value={"ok": True}), + ) + + await send_telegram_notification("chat-id", "hello") + finally: + settings.lnbits_telegram_notifications_access_token = original_token + + wrapper_mock.assert_awaited_once_with("wrapper-token", "chat-id", "hello") + + +@pytest.mark.anyio +async def test_send_email_notification_and_send_email( + settings: Settings, mocker: MockerFixture +): + original_email_enabled = settings.lnbits_email_notifications_enabled + try: + settings.lnbits_email_notifications_enabled = False + disabled = await send_email_notification(["alice@example.com"], "hello") + assert disabled["status"] == "error" + + settings.lnbits_email_notifications_enabled = True + send_email_mock = mocker.patch( + "lnbits.core.services.notifications.send_email", + mocker.AsyncMock(return_value=True), + ) + enabled = await send_email_notification(["alice@example.com"], "hello") + assert enabled == {"status": "ok"} + send_email_mock.assert_awaited_once() + finally: + settings.lnbits_email_notifications_enabled = original_email_enabled + + smtp_server = MagicMock() + smtp_context = MagicMock() + smtp_context.__enter__.return_value = smtp_server + smtp_context.__exit__.return_value = None + mocker.patch( + "lnbits.core.services.notifications.smtplib.SMTP", + return_value=smtp_context, + ) + + assert ( + await send_email( + "smtp.example.com", + 587, + "", + "password", + "from@example.com", + ["to@example.com"], + "Subject", + "Body", + ) + is True + ) + smtp_server.starttls.assert_called_once() + smtp_server.login.assert_called_once_with("from@example.com", "password") + smtp_server.sendmail.assert_called_once() + + with pytest.raises(ValueError, match="Invalid from email address"): + await send_email( + "smtp.example.com", + 587, + "user", + "password", + "bad-email", + ["to@example.com"], + "Subject", + "Body", + ) + + with pytest.raises(ValueError, match="No email addresses provided"): + await send_email( + "smtp.example.com", + 587, + "user", + "password", + "from@example.com", + [], + "Subject", + "Body", + ) + + +@pytest.mark.anyio +async def test_dispatch_webhook_marks_missing_invalid_and_failed_requests( + mocker: MockerFixture, +): + wallet = await _create_wallet() + + payment = await _create_payment(wallet, webhook=None) + await dispatch_webhook(payment) + assert (await get_payment(payment.checking_id)).webhook_status == "-1" + + invalid_payment = await _create_payment(wallet, webhook="https://invalid.example") + invalid_client = MockHTTPClient( + post_response=httpx.Response( + 200, + request=httpx.Request("POST", invalid_payment.webhook), + json={"ok": True}, + ) + ) + mocker.patch( + "lnbits.core.services.notifications.check_callback_url", + side_effect=ValueError("blocked"), + ) + mocker.patch( + "lnbits.core.services.notifications.httpx.AsyncClient", + return_value=invalid_client, + ) + + await dispatch_webhook(invalid_payment) + assert (await get_payment(invalid_payment.checking_id)).webhook_status in {"-1", "200"} + + error_payment = await _create_payment(wallet, webhook="https://error.example") + mocker.patch( + "lnbits.core.services.notifications.check_callback_url", + return_value=None, + ) + mocker.patch( + "lnbits.core.services.notifications.httpx.AsyncClient", + return_value=MockHTTPClient( + post_response=httpx.Response( + 500, + request=httpx.Request("POST", error_payment.webhook), + ) + ), + ) + + await dispatch_webhook(error_payment) + assert (await get_payment(error_payment.checking_id)).webhook_status == "500" + + request_payment = await _create_payment(wallet, webhook="https://request.example") + mocker.patch( + "lnbits.core.services.notifications.httpx.AsyncClient", + return_value=MockHTTPClient( + post_exception=httpx.RequestError( + "boom", + request=httpx.Request("POST", request_payment.webhook), + ) + ), + ) + + await dispatch_webhook(request_payment) + assert (await get_payment(request_payment.checking_id)).webhook_status == "-1" + + +@pytest.mark.anyio +async def test_send_payment_notification_fans_out_to_shared_wallet_and_webhook( + mocker: MockerFixture, +): + wallet = await _create_wallet(name="Primary Wallet") + shared_wallet = await _create_wallet(name="Shared Wallet") + wallet.extra.shared_with = [ + WalletSharePermission( + request_id="share-1", + username="bob", + shared_with_wallet_id=shared_wallet.id, + permissions=[WalletPermission.VIEW_PAYMENTS], + status=WalletShareStatus.APPROVED, + ) + ] + await update_wallet(wallet) + payment = await _create_payment(wallet, webhook="https://webhook.example") + ws_mock = mocker.patch( + "lnbits.core.services.notifications.send_ws_payment_notification", + mocker.AsyncMock(), + ) + chat_mock = mocker.patch( + "lnbits.core.services.notifications.send_chat_payment_notification", + mocker.AsyncMock(), + ) + push_mock = mocker.patch( + "lnbits.core.services.notifications.send_payment_push_notification", + mocker.AsyncMock(), + ) + dispatch_mock = mocker.patch( + "lnbits.core.services.notifications.dispatch_webhook", + mocker.AsyncMock(), + ) + + await send_payment_notification(wallet, payment) + + assert [call.args[0].id for call in ws_mock.await_args_list] == [ + wallet.id, + shared_wallet.id, + ] + chat_mock.assert_awaited_once_with(wallet, payment) + push_mock.assert_awaited_once_with(wallet, payment) + dispatch_mock.assert_awaited_once_with(payment) + + +@pytest.mark.anyio +async def test_send_ws_payment_notification_and_chat_notifications( + settings: Settings, mocker: MockerFixture +): + user_notifications = UserNotifications( + telegram_chat_id="chat-id", + nostr_identifier="alice@example.com", + email_address="alice@example.com", + incoming_payments_sats=1, + outgoing_payments_sats=1, + ) + wallet = await _create_wallet(user_notifications) + payment = await _create_payment( + wallet, + amount_msat=-2_000, + extra={"wallet_fiat_currency": "USD", "wallet_fiat_amount": 5.25}, + ) + websocket_mock = mocker.patch( + "lnbits.core.services.notifications.websocket_manager.send", + mocker.AsyncMock(), + ) + + await send_ws_payment_notification(wallet, payment) + + assert [call.args[0] for call in websocket_mock.await_args_list] == [ + wallet.inkey, + wallet.adminkey, + payment.payment_hash, + ] + + original_outgoing = settings.lnbits_notification_outgoing_payment_amount_sats + original_incoming = settings.lnbits_notification_incoming_payment_amount_sats + try: + settings.lnbits_notification_outgoing_payment_amount_sats = 1 + settings.lnbits_notification_incoming_payment_amount_sats = 1 + admin_mock = mocker.patch( + "lnbits.core.services.notifications.enqueue_admin_notification" + ) + user_mock = mocker.patch( + "lnbits.core.services.notifications.enqueue_user_notification" + ) + + await send_chat_payment_notification(wallet, payment) + finally: + settings.lnbits_notification_outgoing_payment_amount_sats = original_outgoing + settings.lnbits_notification_incoming_payment_amount_sats = original_incoming + + assert admin_mock.call_args.args[0] == NotificationType.outgoing_payment + assert "`5.25`*USD* / " in admin_mock.call_args.args[1]["fiat_value_fmt"] + assert user_mock.call_args.args[0] == NotificationType.outgoing_payment + + +@pytest.mark.anyio +async def test_send_payment_push_notification_and_cleanup_gone_subscriptions( + settings: Settings, mocker: MockerFixture +): + wallet = await _create_wallet() + payment = await _create_payment(wallet, amount_msat=2_000, memo="Thanks") + endpoint = f"https://push.example/{uuid4().hex}" + subscription = await create_webpush_subscription( + endpoint, + wallet.user, + '{"endpoint":"https://push.example"}', + "push.example", + ) + send_push_mock = mocker.patch( + "lnbits.core.services.notifications.send_push_notification", + mocker.AsyncMock(), + ) + + await send_payment_push_notification(wallet, payment) + + assert send_push_mock.await_args.args[0].endpoint == subscription.endpoint + assert send_push_mock.await_args.args[1] == f"LNbits: {wallet.name}" + assert "received 2 sats" in send_push_mock.await_args.args[2] + assert send_push_mock.await_args.args[3] == ( + f"https://{subscription.host}/wallet?usr={wallet.user}&wal={wallet.id}" + ) + + original_privkey = settings.lnbits_webpush_privkey + try: + settings.lnbits_webpush_privkey = "" + exc = WebPushException("gone") + exc.response = SimpleNamespace(status_code=HTTPStatus.GONE, text="gone") + mocker.patch( + "lnbits.core.services.notifications.webpush", + side_effect=exc, + ) + + await send_push_notification(subscription, "Title", "Body") + finally: + settings.lnbits_webpush_privkey = original_privkey + + assert await get_webpush_subscription(subscription.endpoint, wallet.user) is None diff --git a/tests/unit/test_services_payments.py b/tests/unit/test_services_payments.py new file mode 100644 index 000000000..e93d8de3c --- /dev/null +++ b/tests/unit/test_services_payments.py @@ -0,0 +1,450 @@ +from datetime import datetime, timedelta, timezone +from types import SimpleNamespace +from uuid import uuid4 + +import pytest +from pytest_mock.plugin import MockerFixture + +from lnbits.core.crud import ( + create_account, + create_payment, + create_wallet, + get_payment, + get_payments, + update_payment, +) +from lnbits.core.models import ( + Account, + CreateInvoice, + CreatePayment, + PaymentDailyStats, + PaymentState, + Wallet, +) +from lnbits.core.services.payments import ( + cancel_hold_invoice, + calculate_fiat_amounts, + check_payment_status, + check_pending_payments, + check_time_limit_between_transactions, + check_transaction_status, + check_wallet_limits, + create_payment_request, + get_payments_daily_stats, + settle_hold_invoice, + update_pending_payment, + update_pending_payments, + update_wallet_balance, +) +from lnbits.db import Filters +from lnbits.exceptions import InvoiceError, PaymentError +from lnbits.settings import Settings +from lnbits.wallets.base import ( + InvoiceResponse, + PaymentFailedStatus, + PaymentPendingStatus, + PaymentSuccessStatus, +) + + +def _account() -> Account: + account_id = uuid4().hex + return Account(id=account_id, username=f"user_{account_id[:8]}") + + +async def _create_wallet() -> Wallet: + account = _account() + await create_account(account) + return await create_wallet(user_id=account.id, wallet_name=f"wallet_{account.id[:8]}") + + +async def _create_payment( + wallet: Wallet, + *, + amount_msat: int = 2_000, + status: PaymentState = PaymentState.PENDING, + checking_id: str | None = None, + payment_hash: str | None = None, + fee: int = 0, + time: datetime | None = None, +) -> str: + checking_id = checking_id or f"checking_{uuid4().hex[:8]}" + payment_hash = payment_hash or uuid4().hex + payment = await create_payment( + checking_id=checking_id, + data=CreatePayment( + wallet_id=wallet.id, + payment_hash=payment_hash, + bolt11=f"bolt11-{checking_id}", + amount_msat=amount_msat, + memo="memo", + fee=fee, + ), + status=status, + ) + if time: + payment.time = time + payment.created_at = time + payment.updated_at = time + await update_payment(payment) + return checking_id + + +@pytest.mark.anyio +async def test_create_payment_request_routes_by_invoice_type(mocker: MockerFixture): + wallet_payment = SimpleNamespace(checking_id="wallet") + fiat_payment = SimpleNamespace(checking_id="fiat") + wallet_mock = mocker.patch( + "lnbits.core.services.payments.create_wallet_invoice", + mocker.AsyncMock(return_value=wallet_payment), + ) + fiat_mock = mocker.patch( + "lnbits.core.services.payments.create_fiat_invoice", + mocker.AsyncMock(return_value=fiat_payment), + ) + + assert await create_payment_request("wallet-1", CreateInvoice(amount=1)) == wallet_payment + assert ( + await create_payment_request( + "wallet-1", + CreateInvoice(amount=1, fiat_provider="stripe"), + ) + == fiat_payment + ) + wallet_mock.assert_awaited_once() + fiat_mock.assert_awaited_once() + + +@pytest.mark.anyio +async def test_update_pending_payment_and_bulk_pending_updates(mocker: MockerFixture): + wallet = await _create_wallet() + failed_id = await _create_payment(wallet) + success_id = await _create_payment(wallet) + failed_payment = await get_payment(failed_id) + success_payment = await get_payment(success_id) + + mocker.patch( + "lnbits.core.services.payments.check_payment_status", + mocker.AsyncMock(side_effect=[PaymentFailedStatus(), PaymentSuccessStatus()]), + ) + + await update_pending_payment(failed_payment) + await update_pending_payment(success_payment) + + assert (await get_payment(failed_id)).status == PaymentState.FAILED + assert (await get_payment(success_id)).status == PaymentState.SUCCESS + + bulk_wallet = await _create_wallet() + bulk_failed_id = await _create_payment(bulk_wallet) + bulk_success_id = await _create_payment(bulk_wallet) + mocker.patch( + "lnbits.core.services.payments.check_payment_status", + mocker.AsyncMock(side_effect=[PaymentFailedStatus(), PaymentSuccessStatus()]), + ) + + await update_pending_payments(bulk_wallet.id) + + bulk_statuses = { + (await get_payment(bulk_failed_id)).status, + (await get_payment(bulk_success_id)).status, + } + assert bulk_statuses == {PaymentState.FAILED, PaymentState.SUCCESS} + + +@pytest.mark.anyio +async def test_check_pending_payments_skips_voidwallet_and_updates_recent_items( + mocker: MockerFixture, +): + class VoidWallet: + pass + + class FakeWalletSource: + pass + + mocker.patch( + "lnbits.core.services.payments.get_funding_source", + return_value=VoidWallet(), + ) + sleep_mock = mocker.patch("lnbits.core.services.payments.asyncio.sleep", mocker.AsyncMock()) + + await check_pending_payments() + sleep_mock.assert_not_awaited() + + existing_pending = await get_payments(pending=True, exclude_uncheckable=True) + wallet = await _create_wallet() + checking_id = await _create_payment(wallet) + mocker.patch( + "lnbits.core.services.payments.get_funding_source", + return_value=FakeWalletSource(), + ) + mocker.patch( + "lnbits.core.services.payments.check_payment_status", + mocker.AsyncMock(return_value=PaymentSuccessStatus()), + ) + + try: + await check_pending_payments() + finally: + for payment in existing_pending: + payment.status = PaymentState.PENDING + await update_payment(payment) + + assert (await get_payment(checking_id)).status == PaymentState.SUCCESS + assert sleep_mock.await_count >= 1 + + +@pytest.mark.anyio +async def test_update_wallet_balance_validates_credit_and_debit( + settings: Settings, mocker: MockerFixture +): + wallet = await _create_wallet() + wallet.balance_msat = 20_000 + + with pytest.raises(ValueError, match="Amount cannot be 0."): + await update_wallet_balance(wallet, 0) + + with pytest.raises(ValueError, match="can not go into negative balance"): + await update_wallet_balance(wallet, -30) + + payment_secret = (uuid4().hex * 2)[:64] + payment_hash = (uuid4().hex * 2)[:64] + mocker.patch( + "lnbits.core.services.payments.random_secret_and_hash", + return_value=(payment_secret, payment_hash), + ) + mocker.patch( + "lnbits.core.services.payments.fake_privkey", + return_value="privkey", + ) + mocker.patch( + "lnbits.core.services.payments.bolt11_encode", + return_value="encoded-bolt11", + ) + + await update_wallet_balance(wallet, -10) + + debit_payment = await get_payment("internal_" + payment_hash) + assert debit_payment is not None + assert debit_payment.amount == -10_000 + assert debit_payment.status == PaymentState.SUCCESS + + original_max_balance = settings.lnbits_wallet_limit_max_balance + try: + settings.lnbits_wallet_limit_max_balance = 21 + with pytest.raises(ValueError, match="amount exceeds maximum balance"): + await update_wallet_balance(wallet, 5) + + settings.lnbits_wallet_limit_max_balance = 0 + queue_mock = mocker.patch( + "lnbits.core.services.payments.internal_invoice_queue_put", + mocker.AsyncMock(), + ) + + await update_wallet_balance(wallet, 5) + finally: + settings.lnbits_wallet_limit_max_balance = original_max_balance + + credit_payments = [ + payment + for payment in await get_payments(wallet_id=wallet.id, incoming=True) + if payment.memo == "Admin credit" + ] + assert credit_payments + assert credit_payments[0].status == PaymentState.SUCCESS + queue_mock.assert_awaited_once_with(credit_payments[0].checking_id) + + +@pytest.mark.anyio +async def test_check_wallet_limits_and_time_limit( + settings: Settings, mocker: MockerFixture +): + time_limit_mock = mocker.patch( + "lnbits.core.services.payments.check_time_limit_between_transactions", + mocker.AsyncMock(), + ) + daily_limit_mock = mocker.patch( + "lnbits.core.services.payments.check_wallet_daily_withdraw_limit", + mocker.AsyncMock(), + ) + + await check_wallet_limits("wallet-1", 1_000) + + time_limit_mock.assert_awaited_once_with("wallet-1", None) + daily_limit_mock.assert_awaited_once_with("wallet-1", 1_000, None) + + wallet = await _create_wallet() + await _create_payment(wallet, amount_msat=-2_000) + original_limit = settings.lnbits_wallet_limit_secs_between_trans + try: + settings.lnbits_wallet_limit_secs_between_trans = 30 + with pytest.raises(PaymentError) as exc_info: + await check_time_limit_between_transactions(wallet.id) + assert "30 seconds between payments" in exc_info.value.message + + other_wallet = await _create_wallet() + assert await check_time_limit_between_transactions(other_wallet.id) is None + finally: + settings.lnbits_wallet_limit_secs_between_trans = original_limit + + +@pytest.mark.anyio +async def test_calculate_fiat_amounts_handles_conversion_and_errors( + mocker: MockerFixture, +): + wallet = await _create_wallet() + wallet.currency = "EUR" + mocker.patch( + "lnbits.core.services.payments.fiat_amount_as_satoshis", + mocker.AsyncMock(return_value=200), + ) + sat_to_fiat_mock = mocker.patch( + "lnbits.core.services.payments.satoshis_amount_as_fiat", + mocker.AsyncMock(return_value=1.5), + ) + + amount_sat, fiat_amounts = await calculate_fiat_amounts(2.0, wallet, "USD") + + assert amount_sat == 200 + assert fiat_amounts["fiat_currency"] == "USD" + assert fiat_amounts["wallet_fiat_currency"] == "EUR" + assert fiat_amounts["wallet_fiat_amount"] == 1.5 + + sat_to_fiat_mock.side_effect = Exception("boom") + amount_sat, fiat_amounts = await calculate_fiat_amounts(10, wallet, "sat", extra={}) + + assert amount_sat == 10 + assert fiat_amounts == {} + + +@pytest.mark.anyio +async def test_check_transaction_status_and_payment_status(mocker: MockerFixture): + wallet = await _create_wallet() + missing_hash = uuid4().hex + assert (await check_transaction_status(wallet.id, missing_hash)).pending is True + + success_hash = uuid4().hex + success_id = await _create_payment( + wallet, + status=PaymentState.SUCCESS, + payment_hash=success_hash, + fee=-123, + ) + success_status = await check_transaction_status(wallet.id, success_hash) + assert success_status.success is True + assert success_status.fee_msat == -123 + + pending_hash = uuid4().hex + await _create_payment(wallet, payment_hash=pending_hash) + mocker.patch( + "lnbits.core.services.payments.check_payment_status", + mocker.AsyncMock(return_value=PaymentFailedStatus()), + ) + assert (await check_transaction_status(wallet.id, pending_hash)).failed is True + + internal_success = await get_payment(success_id) + internal_success.checking_id = "internal_" + internal_success.payment_hash + internal_success.status = PaymentState.SUCCESS.value + assert (await check_payment_status(internal_success)).success is True + + internal_failed = await get_payment(success_id) + internal_failed.checking_id = "internal_" + internal_failed.payment_hash + internal_failed.status = PaymentState.FAILED.value + assert (await check_payment_status(internal_failed)).failed is True + + internal_fiat = await get_payment(success_id) + internal_fiat.checking_id = "fiat_" + internal_fiat.payment_hash + internal_fiat.status = PaymentState.PENDING.value + internal_fiat.fiat_provider = "stripe" + mocker.patch( + "lnbits.core.services.payments.check_fiat_status", + mocker.AsyncMock(return_value=SimpleNamespace(paid=True)), + ) + assert (await check_payment_status(internal_fiat)).success is True + + outgoing = await get_payment(success_id) + outgoing.checking_id = "external-out" + outgoing.amount = -2_000 + incoming = await get_payment(success_id) + incoming.checking_id = "external-in" + incoming.amount = 2_000 + funding_source = SimpleNamespace( + get_payment_status=mocker.AsyncMock(return_value=PaymentSuccessStatus()), + get_invoice_status=mocker.AsyncMock(return_value=PaymentPendingStatus()), + ) + mocker.patch( + "lnbits.core.services.payments.get_funding_source", + return_value=funding_source, + ) + + assert (await check_payment_status(outgoing)).success is True + assert (await check_payment_status(incoming)).pending is True + + +@pytest.mark.anyio +async def test_get_payments_daily_stats_fills_missing_dates(): + wallet = await _create_wallet() + user_id = wallet.user + now = datetime.now(timezone.utc).replace(hour=12, minute=0, second=0, microsecond=0) + await _create_payment( + wallet, + amount_msat=2_000, + status=PaymentState.SUCCESS, + time=now - timedelta(days=2), + ) + await _create_payment( + wallet, + amount_msat=-500, + status=PaymentState.SUCCESS, + fee=100, + time=now, + ) + + stats = await get_payments_daily_stats(Filters(), user_id=user_id) + + assert [point.date.date() for point in stats[-3:]] == [ + (now - timedelta(days=2)).date(), + (now - timedelta(days=1)).date(), + now.date(), + ] + assert [point.balance for point in stats[-3:]] == [2, 2, 1] + assert stats[-1].fee == 0 + + +@pytest.mark.anyio +async def test_settle_and_cancel_hold_invoice_persist_status(mocker: MockerFixture): + wallet = await _create_wallet() + checking_id = await _create_payment(wallet, payment_hash="33" * 32) + payment = await get_payment(checking_id) + funding_source = SimpleNamespace( + settle_hold_invoice=mocker.AsyncMock( + return_value=InvoiceResponse(ok=True, checking_id="settled") + ), + cancel_hold_invoice=mocker.AsyncMock( + return_value=InvoiceResponse(ok=True, checking_id="cancelled") + ), + ) + mocker.patch( + "lnbits.core.services.payments.get_funding_source", + return_value=funding_source, + ) + mocker.patch( + "lnbits.core.services.payments.verify_preimage", + return_value=False, + ) + + with pytest.raises(InvoiceError, match="Invalid preimage."): + await settle_hold_invoice(payment, "00" * 32) + + mocker.patch( + "lnbits.core.services.payments.verify_preimage", + return_value=True, + ) + + assert (await settle_hold_invoice(payment, "11" * 32)).ok is True + assert (await cancel_hold_invoice(payment)).ok is True + + stored = await get_payment(checking_id) + assert stored.preimage == "11" * 32 + assert stored.extra["hold_invoice_settled"] is True + assert stored.extra["hold_invoice_cancelled"] is True + assert stored.status == PaymentState.FAILED diff --git a/tests/unit/test_services_settings.py b/tests/unit/test_services_settings.py index 6a2fe8f25..09861bf8a 100644 --- a/tests/unit/test_services_settings.py +++ b/tests/unit/test_services_settings.py @@ -1,9 +1,9 @@ -from unittest.mock import AsyncMock - import pytest from pydantic import ValidationError from pytest_mock.plugin import MockerFixture +from lnbits.core.crud import create_admin_settings, delete_admin_settings, get_super_settings +from lnbits.core.crud.settings import get_settings_field from lnbits.core.services.settings import ( check_webpush_settings, dict_to_settings, @@ -74,21 +74,40 @@ def test_update_cached_settings_updates_runtime_values(settings: Settings): 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) + previous_settings = await get_super_settings() + previous_private = settings.lnbits_webpush_privkey + previous_public = settings.lnbits_webpush_pubkey + previous_admin_ui = settings.lnbits_admin_ui + await delete_admin_settings() + + settings.lnbits_webpush_privkey = "" + settings.lnbits_webpush_pubkey = None + 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(), - ) + try: + await check_webpush_settings() - await check_webpush_settings() + stored_private = await get_settings_field("lnbits_webpush_privkey") + stored_public = await get_settings_field("lnbits_webpush_pubkey") - assert settings.lnbits_webpush_privkey == "private-key" - assert settings.lnbits_webpush_pubkey == "public-key" - update_admin.assert_awaited_once() + assert settings.lnbits_webpush_privkey == "private-key" + assert settings.lnbits_webpush_pubkey == "public-key" + assert stored_private is not None + assert stored_private.value == "private-key" + assert stored_public is not None + assert stored_public.value == "public-key" + finally: + await delete_admin_settings() + if previous_settings: + await create_admin_settings( + previous_settings.super_user, + previous_settings.dict(exclude={"super_user"}), + ) + update_cached_settings(previous_settings.dict()) + settings.lnbits_webpush_privkey = previous_private + settings.lnbits_webpush_pubkey = previous_public + settings.lnbits_admin_ui = previous_admin_ui @pytest.mark.anyio @@ -110,15 +129,19 @@ async def test_check_webpush_settings_requires_public_key( 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") + previous_private = settings.lnbits_webpush_privkey + previous_public = settings.lnbits_webpush_pubkey + previous_private_field = await get_settings_field("lnbits_webpush_privkey") + previous_public_field = await get_settings_field("lnbits_webpush_pubkey") + settings.lnbits_webpush_privkey = "existing-private-key" + 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() + try: + await check_webpush_settings() + finally: + settings.lnbits_webpush_privkey = previous_private + settings.lnbits_webpush_pubkey = previous_public + assert await get_settings_field("lnbits_webpush_privkey") == previous_private_field + assert await get_settings_field("lnbits_webpush_pubkey") == previous_public_field vapid.assert_not_called() - update_admin.assert_not_awaited() diff --git a/tests/unit/test_services_users.py b/tests/unit/test_services_users.py new file mode 100644 index 000000000..600dab4aa --- /dev/null +++ b/tests/unit/test_services_users.py @@ -0,0 +1,395 @@ +from uuid import uuid4 + +import pytest + +from lnbits.core.crud import ( + create_account, + create_admin_settings, + create_user_extension, + delete_admin_settings, + get_account, + get_super_settings, + get_user_extensions, + get_wallets, +) +from lnbits.core.crud.settings import get_settings_field, set_settings_field +from lnbits.core.models import Account +from lnbits.core.models.extensions import UserExtension +from lnbits.core.models.users import RegisterUser, UserExtra +from lnbits.core.services.users import ( + check_admin_settings, + check_register_activation_settings, + create_user_account, + create_user_account_no_ckeck, + init_admin_settings, + update_user_account, + update_user_extensions, +) +from lnbits.core.services.settings import update_cached_settings +from lnbits.settings import Settings + + +def _pubkey(value: int) -> str: + return f"{value:064x}" + + +def _account( + *, + id_: str | None = None, + username: str | None = None, + email: str | None = None, + pubkey: str | None = None, +) -> Account: + account_id = id_ or uuid4().hex + return Account( + id=account_id, + username=username or f"user_{account_id[:8]}", + email=email or f"{account_id[:8]}@example.com", + pubkey=pubkey, + ) + + +@pytest.mark.anyio +async def test_create_user_account_rejects_when_registration_disabled( + settings: Settings, +): + settings.lnbits_allow_new_accounts = False + + with pytest.raises(ValueError, match="Account creation is disabled."): + await create_user_account() + + +@pytest.mark.anyio +@pytest.mark.parametrize( + ("existing_data", "new_data", "message"), + [ + ( + {"username": f"user_{uuid4().hex[:8]}"}, + {"username": lambda existing: existing.username}, + "Username already exists.", + ), + ( + {"email": f"{uuid4().hex[:8]}@example.com"}, + {"email": lambda existing: existing.email}, + "Email already exists.", + ), + ( + {"pubkey": _pubkey(1)}, + {"pubkey": lambda existing: existing.pubkey}, + "Pubkey already exists.", + ), + ], +) +async def test_create_user_account_no_check_rejects_duplicate_identity_fields( + existing_data: dict, new_data: dict, message: str +): + existing = _account(**existing_data) + await create_account(existing) + + resolved = { + key: (value(existing) if callable(value) else value) + for key, value in new_data.items() + } + account = _account(**resolved) + + with pytest.raises(ValueError, match=message): + await create_user_account_no_ckeck(account) + + +@pytest.mark.anyio +async def test_create_user_account_no_check_creates_wallet_and_extensions( + settings: Settings, +): + account = _account() + original_default_exts = list(settings.lnbits_user_default_extensions) + try: + settings.lnbits_user_default_extensions = ["default-ext"] + + user = await create_user_account_no_ckeck( + account, + wallet_name="Primary", + default_exts=["extra-ext"], + ) + finally: + settings.lnbits_user_default_extensions = original_default_exts + + wallets = await get_wallets(user.id) + user_extensions = await get_user_extensions(user.id) + + assert len(wallets) == 1 + assert wallets[0].name == "Primary" + assert {ext.extension for ext in user_extensions} == {"default-ext", "extra-ext"} + assert all(ext.active is True for ext in user_extensions) + + +@pytest.mark.anyio +async def test_create_user_account_no_check_ignores_duplicate_extension_insert( + settings: Settings, +): + account = _account() + original_default_exts = list(settings.lnbits_user_default_extensions) + try: + settings.lnbits_user_default_extensions = ["dup-ext"] + + user = await create_user_account_no_ckeck(account, default_exts=["dup-ext"]) + finally: + settings.lnbits_user_default_extensions = original_default_exts + + user_extensions = await get_user_extensions(user.id) + assert [ext.extension for ext in user_extensions] == ["dup-ext"] + + +@pytest.mark.anyio +async def test_update_user_account_requires_existing_user(): + account = _account() + + with pytest.raises(ValueError, match="User does not exist."): + await update_user_account(account) + + +@pytest.mark.anyio +async def test_update_user_account_rejects_conflicting_identity_fields(): + existing = _account(pubkey=_pubkey(2)) + conflict = _account(pubkey=_pubkey(3)) + await create_account(existing) + await create_account(conflict) + + with pytest.raises(ValueError, match="Username already exists."): + await update_user_account( + _account( + id_=existing.id, + username=conflict.username, + email=existing.email, + pubkey=existing.pubkey, + ) + ) + + with pytest.raises(ValueError, match="Email already exists."): + await update_user_account( + _account( + id_=existing.id, + username=existing.username, + email=conflict.email, + pubkey=existing.pubkey, + ) + ) + + with pytest.raises(ValueError, match="Pubkey already exists."): + await update_user_account( + _account( + id_=existing.id, + username=existing.username, + email=existing.email, + pubkey=conflict.pubkey, + ) + ) + + +@pytest.mark.anyio +async def test_update_user_account_updates_persisting_password(): + account = _account(pubkey=_pubkey(4)) + account.hash_password("secret1234") + await create_account(account) + + updated = _account( + id_=account.id, + username=f"updated_{account.id[:8]}", + email=f"{account.id[:8]}+updated@example.com", + pubkey=(uuid4().hex * 2)[:64], + ) + result = await update_user_account(updated) + stored = await get_account(account.id) + + assert result.id == account.id + assert stored is not None + assert stored.username == updated.username + assert stored.email == updated.email + assert stored.pubkey == updated.pubkey + assert stored.password_hash == account.password_hash + + +@pytest.mark.anyio +async def test_update_user_extensions_toggles_existing_and_creates_missing( + settings: Settings, +): + original_default_exts = list(settings.lnbits_user_default_extensions) + try: + settings.lnbits_user_default_extensions = [] + user = await create_user_account(_account()) + finally: + settings.lnbits_user_default_extensions = original_default_exts + + await create_user_extension(UserExtension(user=user.id, extension="keep", active=True)) + await create_user_extension( + UserExtension(user=user.id, extension="enable", active=False) + ) + await create_user_extension( + UserExtension(user=user.id, extension="disable", active=True) + ) + + await update_user_extensions(user.id, ["keep", "enable", "new-ext"]) + + user_extensions = { + ext.extension: ext.active for ext in await get_user_extensions(user.id) + } + assert user_extensions == { + "keep": True, + "enable": True, + "disable": False, + "new-ext": True, + } + + +@pytest.mark.anyio +async def test_check_admin_settings_initializes_cache_and_marks_first_install( + settings: Settings, tmp_path +): + previous_settings = await get_super_settings() + previous_super_user = settings.super_user + previous_data_folder = settings.lnbits_data_folder + previous_admin_ui = settings.lnbits_admin_ui + previous_first_install = settings.first_install + super_user = uuid4().hex + + try: + await delete_admin_settings() + settings.super_user = super_user + settings.lnbits_data_folder = str(tmp_path) + settings.lnbits_admin_ui = True + settings.first_install = False + + await check_admin_settings() + + stored_settings = await get_super_settings() + stored_account = await get_account(super_user) + assert stored_settings is not None + assert stored_settings.super_user == super_user + assert stored_account is not None + assert stored_account.extra.provider == "env" + assert settings.first_install is True + assert (tmp_path / ".super_user").read_text() == super_user + finally: + await delete_admin_settings() + if previous_settings: + await create_admin_settings( + previous_settings.super_user, + previous_settings.dict(exclude={"super_user"}), + ) + update_cached_settings(previous_settings.dict()) + settings.super_user = previous_super_user + settings.lnbits_data_folder = previous_data_folder + settings.lnbits_admin_ui = previous_admin_ui + settings.first_install = previous_first_install + + +@pytest.mark.anyio +async def test_init_admin_settings_creates_account_and_wallet_when_missing(): + super_user = uuid4().hex + + result = await init_admin_settings(super_user) + + wallets = await get_wallets(super_user) + assert result.super_user == super_user + assert await get_account(super_user) is not None + assert len(wallets) == 1 + + +@pytest.mark.anyio +async def test_check_register_activation_settings_handles_invitation_codes( + settings: Settings, +): + reusable = "reusable-code" + one_time = "one-time-code" + original_require_activation = settings.lnbits_require_user_activation + original_by_invite = settings.lnbits_user_activation_by_invitation_code + original_reusable = settings.lnbits_register_reusable_activation_code + original_one_time = list(settings.lnbits_register_one_time_activation_codes) + previous_stored_codes = await get_settings_field( + "lnbits_register_one_time_activation_codes" + ) + + try: + settings.lnbits_require_user_activation = False + assert ( + await check_register_activation_settings( + RegisterUser( + username=f"user_{uuid4().hex[:8]}", + password="secret1234", + password_repeat="secret1234", + ) + ) + is None + ) + + settings.lnbits_require_user_activation = True + settings.lnbits_user_activation_by_invitation_code = True + settings.lnbits_register_reusable_activation_code = reusable + settings.lnbits_register_one_time_activation_codes = [one_time] + + with pytest.raises(ValueError, match="Invitation code cannot be empty."): + await check_register_activation_settings( + RegisterUser( + username=f"user_{uuid4().hex[:8]}", + password="secret1234", + password_repeat="secret1234", + invitation_code=" ", + ) + ) + + assert ( + await check_register_activation_settings( + RegisterUser( + username=f"user_{uuid4().hex[:8]}", + password="secret1234", + password_repeat="secret1234", + invitation_code=reusable, + ) + ) + is None + ) + + assert ( + await check_register_activation_settings( + RegisterUser( + username=f"user_{uuid4().hex[:8]}", + password="secret1234", + password_repeat="secret1234", + invitation_code=one_time, + ) + ) + is None + ) + assert one_time not in settings.lnbits_register_one_time_activation_codes + stored_codes = await get_settings_field("lnbits_register_one_time_activation_codes") + assert stored_codes is not None + assert stored_codes.value == [] + + with pytest.raises(ValueError, match="Invalid invitation code."): + await check_register_activation_settings( + RegisterUser( + username=f"user_{uuid4().hex[:8]}", + password="secret1234", + password_repeat="secret1234", + invitation_code="bad-code", + ) + ) + + settings.lnbits_user_activation_by_invitation_code = False + with pytest.raises(ValueError, match="No activation method provided."): + await check_register_activation_settings( + RegisterUser( + username=f"user_{uuid4().hex[:8]}", + password="secret1234", + password_repeat="secret1234", + invitation_code=reusable, + ) + ) + finally: + settings.lnbits_require_user_activation = original_require_activation + settings.lnbits_user_activation_by_invitation_code = original_by_invite + settings.lnbits_register_reusable_activation_code = original_reusable + settings.lnbits_register_one_time_activation_codes = original_one_time + await set_settings_field( + "lnbits_register_one_time_activation_codes", + previous_stored_codes.value if previous_stored_codes else original_one_time, + ) diff --git a/tests/unit/test_services_websockets.py b/tests/unit/test_services_websockets.py new file mode 100644 index 000000000..20a2c1c9d --- /dev/null +++ b/tests/unit/test_services_websockets.py @@ -0,0 +1,76 @@ +import asyncio + +import pytest +from fastapi import WebSocketDisconnect +from pytest_mock.plugin import MockerFixture + +from lnbits.core.services.websockets import ( + WebsocketConnectionManager, + websocket_updater, +) +from lnbits.settings import Settings + + +class FakeWebSocket: + def __init__(self, received=None): + self.received = list(received or []) + self.accepted = False + self.sent: list[str] = [] + + async def accept(self): + self.accepted = True + + async def receive_text(self): + if self.received: + value = self.received.pop(0) + if isinstance(value, Exception): + raise value + return value + raise WebSocketDisconnect() + + async def send_text(self, data: str): + self.sent.append(data) + + +@pytest.mark.anyio +async def test_websocket_connection_manager_connect_and_send(): + manager = WebsocketConnectionManager() + websocket = FakeWebSocket() + + conn = await manager.connect("item-1", websocket) + await manager.send("item-1", "payload") + + assert websocket.accepted is True + assert manager.has_connection("item-1") is True + assert manager.get_connections("item-1") == [conn] + assert websocket.sent == ["payload"] + + +@pytest.mark.anyio +async def test_websocket_connection_manager_listen_queues_messages_and_disconnects( + settings: Settings, +): + manager = WebsocketConnectionManager() + websocket = FakeWebSocket(["hello", WebSocketDisconnect()]) + conn = await manager.connect("item-2", websocket) + original_running = settings.lnbits_running + try: + settings.lnbits_running = True + await manager.listen(conn) + finally: + settings.lnbits_running = original_running + + assert conn.receive_queue.get_nowait() == "hello" + assert manager.has_connection("item-2") is False + + +@pytest.mark.anyio +async def test_websocket_updater_delegates_to_manager(mocker: MockerFixture): + send = mocker.patch( + "lnbits.core.services.websockets.websocket_manager.send", + mocker.AsyncMock(), + ) + + await websocket_updater("item-3", "data") + + send.assert_awaited_once_with("item-3", "data")