diff --git a/tests/api/test_asset_api.py b/tests/api/test_asset_api.py index b6667d45e..7707ad92e 100644 --- a/tests/api/test_asset_api.py +++ b/tests/api/test_asset_api.py @@ -1,39 +1,11 @@ -from io import BytesIO from uuid import uuid4 import pytest -from fastapi import UploadFile from httpx import AsyncClient -from PIL import Image -from starlette.datastructures import Headers from lnbits.core.crud.assets import get_user_asset from lnbits.core.services.assets import create_user_asset - - -def _png_bytes() -> bytes: - image = Image.new("RGB", (32, 32), color="blue") - buffer = BytesIO() - image.save(buffer, format="PNG") - return buffer.getvalue() - - -def _upload_file(contents: bytes, filename: str, content_type: str) -> UploadFile: - return UploadFile( - BytesIO(contents), - filename=filename, - headers=Headers({"content-type": content_type}), - ) - - -async def _user_headers(client: AsyncClient, user_id: str) -> dict[str, str]: - response = await client.post("/api/v1/auth/usr", json={"usr": user_id}) - client.cookies.clear() - access_token = response.json()["access_token"] - return { - "Authorization": f"Bearer {access_token}", - "Content-type": "application/json", - } +from tests.helpers import get_png_bytes, get_user_token_headers, make_upload_file @pytest.mark.anyio @@ -100,10 +72,14 @@ async def test_asset_api_enforces_visibility_and_supports_admin_updates( ): private_asset = await create_user_asset( from_user.id, - _upload_file(_png_bytes(), f"private_{uuid4().hex[:8]}.png", "image/png"), + make_upload_file( + get_png_bytes(), + filename=f"private_{uuid4().hex[:8]}.png", + content_type="image/png", + ), is_public=False, ) - other_user_headers = await _user_headers(client, to_user.id) + other_user_headers = await get_user_token_headers(client, to_user.id) anonymous = await client.get(f"/api/v1/assets/{private_asset.id}/data") assert anonymous.status_code == 404 @@ -152,7 +128,7 @@ async def test_asset_api_validates_uploads_and_missing_assets( stored = await create_user_asset( "missing-user-check", - _upload_file(b"content", "content.txt", "text/plain"), + make_upload_file(b"content", filename="content.txt", content_type="text/plain"), is_public=True, ) fetched = await get_user_asset("missing-user-check", stored.id) diff --git a/tests/api/test_extension_api.py b/tests/api/test_extension_api.py index 9abb886ad..c3d64ffaa 100644 --- a/tests/api/test_extension_api.py +++ b/tests/api/test_extension_api.py @@ -19,7 +19,6 @@ from lnbits.core.models.extensions import ( CreateExtensionReview, Extension, ExtensionConfig, - ExtensionMeta, ExtensionRelease, InstallableExtension, PayToEnableInfo, @@ -49,45 +48,7 @@ from lnbits.core.views.extension_api import ( get_pay_to_enable_invoice, get_pay_to_install_invoice, ) - - -def _release(ext_id: str, version: str = "1.0.0") -> ExtensionRelease: - return ExtensionRelease( - name=ext_id, - version=version, - archive=f"https://example.com/{ext_id}.zip", - source_repo="org/repo", - hash=f"hash-{ext_id}", - details_link=f"https://example.com/{ext_id}/details.json", - repo=f"https://github.com/org/{ext_id}", - icon=f"/{ext_id}/static/icon.png", - pay_link=f"https://pay.example/{ext_id}", - ) - - -def _installable_extension( - ext_id: str, - *, - active: bool = True, - pay_to_enable: PayToEnableInfo | None = None, - dependencies: list[str] | None = None, - payments: list[ReleasePaymentInfo] | None = None, -) -> InstallableExtension: - release = _release(ext_id) - return InstallableExtension( - id=ext_id, - name=f"Extension {ext_id}", - version=release.version, - active=active, - short_description="Demo extension", - icon=release.icon, - meta=ExtensionMeta( - installed_release=release, - pay_to_enable=pay_to_enable, - dependencies=dependencies or [], - payments=payments or [], - ), - ) +from tests.helpers import make_extension_release, make_installable_extension class _MockHTTPResponse: @@ -132,7 +93,7 @@ class _MockHTTPClient: @pytest.mark.anyio async def test_extension_api_install_details_and_release_endpoints(mocker): ext_id = f"ext_{uuid4().hex[:8]}" - release = _release(ext_id) + release = make_extension_release(ext_id) create_data = CreateExtension( ext_id=ext_id, archive=release.archive, @@ -172,7 +133,7 @@ async def test_extension_api_install_details_and_release_endpoints(mocker): assert details["icon"] == release.icon assert details["repo"] == release.repo - installed_ext = _installable_extension( + installed_ext = make_installable_extension( ext_id, payments=[ ReleasePaymentInfo( @@ -218,7 +179,7 @@ async def test_extension_api_pay_to_enable_and_catalog_views(mocker, admin_user) ext_id = f"paid_{uuid4().hex[:8]}" await create_installed_extension( - _installable_extension( + make_installable_extension( ext_id, pay_to_enable=PayToEnableInfo( required=True, amount=10, wallet=admin_wallet.id @@ -294,7 +255,7 @@ async def test_extension_api_pay_to_enable_and_catalog_views(mocker, admin_user) visible_extensions = await api_get_user_extensions(AccountId(id=regular_user.id)) assert [ext.code for ext in visible_extensions] == [ext_id] - catalog_entry = _installable_extension( + catalog_entry = make_installable_extension( ext_id, pay_to_enable=PayToEnableInfo(required=True, amount=21, wallet=admin_wallet.id), ) @@ -315,11 +276,11 @@ async def test_extension_api_activate_uninstall_install_invoice_and_cleanup(mock uninstall_ext = f"uninstall_{uuid4().hex[:8]}" db_ext = f"db_{uuid4().hex[:8]}" - await create_installed_extension(_installable_extension(base_ext)) + await create_installed_extension(make_installable_extension(base_ext)) await create_installed_extension( - _installable_extension(dependent_ext, dependencies=[base_ext]) + make_installable_extension(dependent_ext, dependencies=[base_ext]) ) - await create_installed_extension(_installable_extension(uninstall_ext)) + await create_installed_extension(make_installable_extension(uninstall_ext)) mocker.patch( "lnbits.core.views.extension_api.get_valid_extensions", @@ -370,7 +331,7 @@ async def test_extension_api_activate_uninstall_install_invoice_and_cleanup(mock install_invoice = await create_wallet_invoice( wallet.id, CreateInvoice(out=False, amount=33, memo="install extension") ) - release = _release(base_ext, version="2.0.0") + release = make_extension_release(base_ext, version="2.0.0") payment_info = ReleasePaymentInfo( amount=33, pay_link=release.pay_link, diff --git a/tests/api/test_extensions_builder_api.py b/tests/api/test_extensions_builder_api.py index d81912d77..80e1cf620 100644 --- a/tests/api/test_extensions_builder_api.py +++ b/tests/api/test_extensions_builder_api.py @@ -7,19 +7,8 @@ from lnbits.core.crud.extensions import create_user_extension, get_user_extensio from lnbits.core.crud.users import get_account from lnbits.core.models.extensions import ( Extension, - ExtensionRelease, UserExtension, ) -from lnbits.core.models.extensions_builder import ( - ActionFields, - ClientDataFields, - DataField, - DataFields, - ExtensionData, - OwnerDataFields, - PublicPageFields, - SettingsFields, -) from lnbits.core.models.users import AccountId from lnbits.core.views.extensions_builder_api import ( api_build_extension, @@ -28,41 +17,7 @@ from lnbits.core.views.extensions_builder_api import ( api_preview_extension, ) from lnbits.settings import Settings - - -def _extension_data(ext_id: str = "demoext") -> ExtensionData: - return ExtensionData( - id=ext_id, - 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(), - ), - ) - - -def _release(ext_id: str) -> ExtensionRelease: - return ExtensionRelease( - name=ext_id, - version="0.1.0", - archive=f"https://example.com/{ext_id}.zip", - source_repo="org/repo", - is_github_release=False, - hash=f"hash-{ext_id}", - icon=f"/{ext_id}/static/image/{ext_id}.png", - ) +from tests.helpers import make_extension_data, make_extension_release @pytest.mark.anyio @@ -70,7 +25,7 @@ async def test_extensions_builder_api_build_preview_and_cleanup( tmp_path, settings: Settings, mocker, from_user ): ext_id = f"builder_{uuid4().hex[:8]}" - data = _extension_data(ext_id) + data = make_extension_data(ext_id) build_dir = tmp_path / "build" build_dir.mkdir(parents=True, exist_ok=True) (build_dir / "index.txt").write_text("hello") @@ -78,7 +33,9 @@ async def test_extensions_builder_api_build_preview_and_cleanup( original_data_folder = settings.lnbits_data_folder build_mock = mocker.patch( "lnbits.core.views.extensions_builder_api.build_extension_from_data", - mocker.AsyncMock(return_value=(_release(ext_id), build_dir)), + mocker.AsyncMock( + return_value=(make_extension_release(ext_id, "0.1.0"), build_dir) + ), ) clean_mock = mocker.patch( "lnbits.core.views.extensions_builder_api.clean_extension_builder_data" @@ -108,7 +65,7 @@ async def test_extensions_builder_api_deploy_updates_user_extension( tmp_path, settings: Settings, mocker, admin_user ): ext_id = f"deploy_{uuid4().hex[:8]}" - data = _extension_data(ext_id) + data = make_extension_data(ext_id) account = await get_account(admin_user.id) assert account is not None @@ -123,7 +80,9 @@ async def test_extensions_builder_api_deploy_updates_user_extension( mocker.patch( "lnbits.core.views.extensions_builder_api.build_extension_from_data", - mocker.AsyncMock(return_value=(_release(ext_id), build_root)), + mocker.AsyncMock( + return_value=(make_extension_release(ext_id, "0.1.0"), build_root) + ), ) install_mock = mocker.patch( "lnbits.core.views.extensions_builder_api.install_extension", diff --git a/tests/api/test_lnurl_api.py b/tests/api/test_lnurl_api.py index 5099e39cb..5cbdba40a 100644 --- a/tests/api/test_lnurl_api.py +++ b/tests/api/test_lnurl_api.py @@ -1,7 +1,6 @@ from uuid import uuid4 import pytest -from bolt11.types import MilliSatoshi from fastapi import HTTPException from lnurl import ( LnAddress, @@ -13,7 +12,7 @@ from lnurl import ( LnurlResponseException, ) from lnurl.models import MessageAction -from lnurl.types import CallbackUrl, LightningInvoice, LnurlPayMetadata +from lnurl.types import CallbackUrl, LightningInvoice from pydantic import parse_obj_as from lnbits.core.models import Account, CreateInvoice @@ -27,6 +26,7 @@ from lnbits.core.views.lnurl_api import ( api_payments_pay_lnurl, api_perform_lnurlauth, ) +from tests.helpers import make_lnurl_pay_response TEST_BOLT11 = ( "lnbc1pnsu5z3pp57getmdaxhg5kc9yh2a2qsh7cjf4gnccgkw0qenm8vsqv50w7s" @@ -37,20 +37,9 @@ TEST_BOLT11 = ( ) -def _pay_response() -> LnurlPayResponse: - return LnurlPayResponse( - callback=parse_obj_as(CallbackUrl, "https://example.com/callback"), - minSendable=MilliSatoshi(1_000), - maxSendable=MilliSatoshi(10_000), - metadata=LnurlPayMetadata( - '[["text/plain","Test payment"],["text/identifier","alice@example.com"]]' - ), - ) - - @pytest.mark.anyio async def test_lnurl_api_scan_routes_validate_and_forward(mocker): - pay_response = _pay_response() + pay_response = make_lnurl_pay_response() mocker.patch( "lnbits.core.views.lnurl_api.lnurl_handle", mocker.AsyncMock(return_value=pay_response), @@ -92,7 +81,7 @@ async def test_lnurl_api_auth_and_pay_flow(mocker): ) wallet = user.wallets[0] wallet_info = WalletTypeInfo(key_type=KeyType.admin, wallet=wallet) - pay_response = _pay_response() + pay_response = make_lnurl_pay_response() payment = await create_wallet_invoice( wallet.id, CreateInvoice(out=False, amount=21, memo="lnurl") ) diff --git a/tests/api/test_payment_api.py b/tests/api/test_payment_api.py index 2b5646848..1371b47cc 100644 --- a/tests/api/test_payment_api.py +++ b/tests/api/test_payment_api.py @@ -33,30 +33,6 @@ ZERO_AMOUNT_INVOICE = ( ) -async def _create_payment( - wallet_id: str, - *, - amount_msat: int, - status: PaymentState = PaymentState.SUCCESS, - payment_hash: str | None = None, - tag: str | None = None, -) -> str: - checking_id = f"checking_{uuid4().hex[:8]}" - await create_payment( - checking_id=checking_id, - data=CreatePayment( - wallet_id=wallet_id, - payment_hash=payment_hash or uuid4().hex, - bolt11=f"bolt11_{checking_id}", - amount_msat=amount_msat, - memo=f"payment_{checking_id}", - extra={"tag": tag} if tag else {}, - ), - status=status, - ) - return checking_id - - @pytest.mark.anyio async def test_payment_api_stats_and_all_paginated(admin_user): first_user = await create_user_account( @@ -185,3 +161,27 @@ async def test_payment_api_fee_reserve_and_hold_invoice_actions(mocker): ) assert cancelled.failed is True cancel_mock.assert_awaited_once() + + +async def _create_payment( + wallet_id: str, + *, + amount_msat: int, + status: PaymentState = PaymentState.SUCCESS, + payment_hash: str | None = None, + tag: str | None = None, +) -> str: + checking_id = f"checking_{uuid4().hex[:8]}" + await create_payment( + checking_id=checking_id, + data=CreatePayment( + wallet_id=wallet_id, + payment_hash=payment_hash or uuid4().hex, + bolt11=f"bolt11_{checking_id}", + amount_msat=amount_msat, + memo=f"payment_{checking_id}", + extra={"tag": tag} if tag else {}, + ), + status=status, + ) + return checking_id diff --git a/tests/api/test_wallet_api.py b/tests/api/test_wallet_api.py index 3293f76d7..9ec2d560d 100644 --- a/tests/api/test_wallet_api.py +++ b/tests/api/test_wallet_api.py @@ -8,10 +8,6 @@ from lnbits.core.models.users import Account from lnbits.core.services.users import create_user_account -def _admin_headers(adminkey: str) -> dict[str, str]: - return {"X-Api-Key": adminkey, "Content-type": "application/json"} - - @pytest.mark.anyio async def test_wallet_api_share_invite_reject_accept_and_delete( http_client: AsyncClient, @@ -188,3 +184,7 @@ async def test_wallet_api_shared_wallet_requires_source_id(http_client: AsyncCli assert ( response.json()["detail"] == "Shared wallet ID is required for shared wallets." ) + + +def _admin_headers(adminkey: str) -> dict[str, str]: + return {"X-Api-Key": adminkey, "Content-type": "application/json"} diff --git a/tests/helpers.py b/tests/helpers.py index a5bc7f16a..3bf7d6521 100644 --- a/tests/helpers.py +++ b/tests/helpers.py @@ -1,8 +1,33 @@ import random import string +from io import BytesIO -from pydantic import BaseModel +from bolt11.types import MilliSatoshi +from fastapi import UploadFile +from httpx import AsyncClient +from lnurl import LnurlPayResponse +from lnurl.types import CallbackUrl, LnurlPayMetadata +from PIL import Image +from pydantic import BaseModel, parse_obj_as +from starlette.datastructures import Headers +from lnbits.core.models.extensions import ( + ExtensionMeta, + ExtensionRelease, + InstallableExtension, + PayToEnableInfo, + ReleasePaymentInfo, +) +from lnbits.core.models.extensions_builder import ( + ActionFields, + ClientDataFields, + DataField, + DataFields, + ExtensionData, + OwnerDataFields, + PublicPageFields, + SettingsFields, +) from lnbits.settings import settings from lnbits.wallets import get_funding_source, set_funding_source @@ -41,6 +66,123 @@ async def get_random_invoice_data(): return {"out": False, "amount": 10, "memo": f"test_memo_{get_random_string(10)}"} +def get_png_bytes(*, color: str = "blue", size: tuple[int, int] = (32, 32)) -> bytes: + image = Image.new("RGB", size, color=color) + buffer = BytesIO() + image.save(buffer, format="PNG") + return buffer.getvalue() + + +def make_upload_file( + contents: bytes, + *, + filename: str, + content_type: str | None, +) -> UploadFile: + headers = ( + Headers({"content-type": content_type}) if content_type is not None else None + ) + return UploadFile(BytesIO(contents), filename=filename, headers=headers) + + +async def get_user_token_headers(client: AsyncClient, user_id: str) -> dict[str, str]: + response = await client.post("/api/v1/auth/usr", json={"usr": user_id}) + client.cookies.clear() + return { + "Authorization": f"Bearer {response.json()['access_token']}", + "Content-type": "application/json", + } + + +def make_extension_data(ext_id: str = "demoext") -> ExtensionData: + return ExtensionData( + id=ext_id, + 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(), + ), + ) + + +def make_extension_release(ext_id: str, version: str = "1.0.0") -> ExtensionRelease: + return ExtensionRelease( + name=ext_id, + version=version, + archive=f"https://example.com/{ext_id}.zip", + source_repo="org/repo", + hash=f"hash-{ext_id}", + details_link=f"https://example.com/{ext_id}/details.json", + repo=f"https://github.com/org/{ext_id}", + icon=f"/{ext_id}/static/icon.png", + pay_link=f"https://pay.example/{ext_id}", + is_github_release=False, + is_version_compatible=True, + ) + + +def make_installable_extension( + ext_id: str, + *, + version: str = "1.0.0", + compatible: bool = True, + active: bool = True, + pay_to_enable: PayToEnableInfo | None = None, + dependencies: list[str] | None = None, + payments: list[ReleasePaymentInfo] | None = None, +) -> InstallableExtension: + release = make_extension_release(ext_id, version) + release.is_version_compatible = compatible + return InstallableExtension( + id=ext_id, + name=f"Extension {ext_id}", + version=version, + active=active, + short_description="Demo extension", + icon=release.icon, + meta=ExtensionMeta( + installed_release=release, + pay_to_enable=pay_to_enable, + dependencies=dependencies or [], + payments=payments or [], + ), + ) + + +def make_lnurl_pay_response( + *, + min_sendable_msat: int = 1_000, + max_sendable_msat: int = 10_000, + text: str = "Test payment", + identifier: str = "alice@example.com", + callback: str = "https://example.com/callback", +) -> LnurlPayResponse: + return LnurlPayResponse( + callback=parse_obj_as(CallbackUrl, callback), + minSendable=MilliSatoshi(min_sendable_msat), + maxSendable=MilliSatoshi(max_sendable_msat), + metadata=LnurlPayMetadata( + f"[[" + f'"text/plain","{text}"' + f"],[" + f'"text/identifier","{identifier}"' + f"]]" + ), + ) + + settings.lnbits_backend_wallet_class = "FakeWallet" set_funding_source("FakeWallet") funding_source = get_funding_source() diff --git a/tests/unit/test_services_assets.py b/tests/unit/test_services_assets.py index 73e2e7484..20751ab6d 100644 --- a/tests/unit/test_services_assets.py +++ b/tests/unit/test_services_assets.py @@ -2,45 +2,26 @@ from io import BytesIO from uuid import uuid4 import pytest -from fastapi import UploadFile from PIL import Image from pytest_mock.plugin import MockerFixture -from starlette.datastructures import Headers 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 - - -def _make_upload_file( - contents: bytes, - *, - filename: str, - content_type: str | None, -) -> UploadFile: - headers = ( - Headers({"content-type": content_type}) if content_type is not None else None - ) - return UploadFile(BytesIO(contents), filename=filename, headers=headers) +from tests.helpers import make_upload_file @pytest.mark.anyio async def test_create_user_asset_validates_upload_constraints( settings: Settings, mocker: MockerFixture ): - file_without_type = _make_upload_file(b"hello", filename="a.txt", content_type=None) + file_without_type = make_upload_file(b"hello", filename="a.txt", content_type=None) 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 = _make_upload_file( + bad_type = make_upload_file( b"hello", filename="bad.bin", content_type="application/x-msdownload", @@ -58,12 +39,12 @@ async def test_create_user_asset_validates_upload_constraints( settings.lnbits_max_asset_size_mb = 1 settings.lnbits_assets_no_limit_users = [] limited_user = await _create_user() - allowed_type = _make_upload_file( + allowed_type = make_upload_file( b"hello", filename="ok.txt", content_type="text/plain" ) await create_user_asset(limited_user, allowed_type, is_public=False) - blocked_by_count = _make_upload_file( + blocked_by_count = make_upload_file( b"again", filename="again.txt", content_type="text/plain", @@ -73,7 +54,7 @@ async def test_create_user_asset_validates_upload_constraints( settings.lnbits_max_asset_size_mb = 0.000001 oversized_user = await _create_user() - large_file = _make_upload_file( + large_file = make_upload_file( b"0123456789", filename="ok.txt", content_type="text/plain", @@ -93,7 +74,7 @@ async def test_create_user_asset_success(mocker: MockerFixture): "lnbits.core.services.assets.thumbnail_from_bytes", return_value=None, ) - file = _make_upload_file(b"hello", filename="hello.txt", content_type="text/plain") + file = make_upload_file(b"hello", filename="hello.txt", content_type="text/plain") asset = await create_user_asset(user_id, file, is_public=True) stored = await get_user_asset(user_id, asset.id) @@ -120,3 +101,9 @@ def test_thumbnail_from_bytes_success_and_failure(): assert thumbnail is not None assert isinstance(thumbnail.getvalue(), bytes) assert thumbnail_from_bytes(b"not-an-image") is None + + +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 diff --git a/tests/unit/test_services_extensions.py b/tests/unit/test_services_extensions.py index eba1941d8..df6cb2d6d 100644 --- a/tests/unit/test_services_extensions.py +++ b/tests/unit/test_services_extensions.py @@ -11,8 +11,6 @@ from lnbits.core.crud import ( ) from lnbits.core.models.extensions import ( Extension, - ExtensionMeta, - ExtensionRelease, InstallableExtension, ReleasePaymentInfo, ) @@ -27,39 +25,14 @@ from lnbits.core.services.extensions import ( 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 [], - ), - ) +from tests.helpers import make_installable_extension @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) + ext_info = make_installable_extension(f"ext_{uuid4().hex[:8]}", compatible=False) original_data_folder = settings.lnbits_data_folder original_extensions_path = settings.lnbits_extensions_path try: @@ -78,7 +51,7 @@ async def test_install_extension_creates_new_extension_and_starts_background_wor tmp_path, settings: Settings, mocker: MockerFixture ): ext_id = f"ext_{uuid4().hex[:8]}" - ext_info = _installable_extension(ext_id) + ext_info = make_installable_extension(ext_id) original_data_folder = settings.lnbits_data_folder original_extensions_path = settings.lnbits_extensions_path download_mock = mocker.patch.object( @@ -125,8 +98,8 @@ async def test_install_extension_updates_existing_upgrade_and_preserves_payments 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") + existing_ext = make_installable_extension(ext_id, payments=[existing_payment]) + updated_ext = make_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") @@ -175,7 +148,7 @@ 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) + ext_info = make_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) @@ -274,8 +247,8 @@ async def test_get_valid_extensions_and_single_extension_respect_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) + ext_one = make_installable_extension(ext_id_one) + ext_two = make_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 diff --git a/tests/unit/test_services_extensions_builder.py b/tests/unit/test_services_extensions_builder.py index ece1c5547..d33770f96 100644 --- a/tests/unit/test_services_extensions_builder.py +++ b/tests/unit/test_services_extensions_builder.py @@ -7,52 +7,20 @@ 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(), - ), - ) +from tests.helpers import make_extension_data @pytest.mark.anyio async def test_build_extension_from_data_orchestrates_builder_steps( tmp_path, mocker: MockerFixture ): - data = _extension_data() + data = make_extension_data() release = ExtensionRelease( name="stub", version="0.1.0", diff --git a/tests/unit/test_services_funding_source.py b/tests/unit/test_services_funding_source.py index 402eda4db..9c2fa568d 100644 --- a/tests/unit/test_services_funding_source.py +++ b/tests/unit/test_services_funding_source.py @@ -18,14 +18,6 @@ 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 @@ -169,3 +161,11 @@ async def test_check_balance_delta_changed_tracks_and_notifies( finally: settings_any.latest_balance_delta_sats = original_latest settings.notification_balance_delta_threshold_sats = original_threshold + + +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 diff --git a/tests/unit/test_services_lnurl.py b/tests/unit/test_services_lnurl.py index de6976898..4145da67e 100644 --- a/tests/unit/test_services_lnurl.py +++ b/tests/unit/test_services_lnurl.py @@ -6,12 +6,11 @@ from lnurl import ( LnAddress, LnurlErrorResponse, LnurlPayActionResponse, - LnurlPayResponse, LnurlResponseException, LnurlSuccessResponse, LnurlWithdrawResponse, ) -from lnurl.types import CallbackUrl, LightningInvoice, LnurlPayMetadata +from lnurl.types import CallbackUrl, LightningInvoice from pydantic import parse_obj_as from pytest_mock.plugin import MockerFixture @@ -25,6 +24,7 @@ from lnbits.core.services.lnurl import ( perform_withdraw, store_paylink, ) +from tests.helpers import make_lnurl_pay_response TEST_BOLT11 = ( "lnbc1pnsu5z3pp57getmdaxhg5kc9yh2a2qsh7cjf4gnccgkw0qenm8vsqv50w7s" @@ -35,33 +35,6 @@ TEST_BOLT11 = ( ) -def _make_pay_response() -> LnurlPayResponse: - return LnurlPayResponse( - callback=parse_obj_as(CallbackUrl, "https://example.com/callback"), - minSendable=MilliSatoshi(1), - maxSendable=MilliSatoshi(10_000), - metadata=LnurlPayMetadata( - '[["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( @@ -105,7 +78,7 @@ async def test_perform_withdraw_rejects_error_response(mocker: MockerFixture): @pytest.mark.anyio async def test_get_pr_from_lnurl_success_and_error(mocker: MockerFixture): - pay_response = _make_pay_response() + pay_response = make_lnurl_pay_response(min_sendable_msat=1, text="Test") mocker.patch( "lnbits.core.services.lnurl.handle", mocker.AsyncMock(return_value=pay_response), @@ -131,7 +104,7 @@ async def test_get_pr_from_lnurl_success_and_error(mocker: MockerFixture): async def test_fetch_lnurl_pay_request_converts_currency_and_stores_paylink( mocker: MockerFixture, ): - pay_response = _make_pay_response() + pay_response = make_lnurl_pay_response(min_sendable_msat=1, text="Test") action_response = LnurlPayActionResponse( pr=LightningInvoice(TEST_BOLT11), disposable=False ) @@ -168,7 +141,7 @@ async def test_fetch_lnurl_pay_request_converts_currency_and_stores_paylink( @pytest.mark.anyio async def test_store_paylink_appends_and_updates_existing(): wallet = await _create_wallet() - pay_response = _make_pay_response() + pay_response = make_lnurl_pay_response(min_sendable_msat=1, text="Test") action_response = LnurlPayActionResponse( pr=LightningInvoice(TEST_BOLT11), disposable=False ) @@ -191,3 +164,19 @@ async def test_store_paylink_appends_and_updates_existing(): 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 + + +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") diff --git a/tests/unit/test_services_notifications.py b/tests/unit/test_services_notifications.py index 8d4424c2a..d379f2478 100644 --- a/tests/unit/test_services_notifications.py +++ b/tests/unit/test_services_notifications.py @@ -69,53 +69,6 @@ class MockHTTPClient: 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 @@ -606,3 +559,50 @@ async def test_send_payment_push_notification_and_cleanup_gone_subscriptions( settings.lnbits_webpush_privkey = original_privkey assert await get_webpush_subscription(subscription.endpoint, wallet.user) is None + + +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) diff --git a/tests/unit/test_services_payments.py b/tests/unit/test_services_payments.py index beb54e68d..0a22b5550 100644 --- a/tests/unit/test_services_payments.py +++ b/tests/unit/test_services_payments.py @@ -46,51 +46,6 @@ from lnbits.wallets.base import ( ) -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") @@ -454,3 +409,48 @@ async def test_settle_and_cancel_hold_invoice_persist_status(mocker: MockerFixtu assert stored.extra["hold_invoice_settled"] is True assert stored.extra["hold_invoice_cancelled"] is True assert stored.status == PaymentState.FAILED + + +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 diff --git a/tests/unit/test_services_users.py b/tests/unit/test_services_users.py index bd809c7f7..765669c5b 100644 --- a/tests/unit/test_services_users.py +++ b/tests/unit/test_services_users.py @@ -29,26 +29,6 @@ from lnbits.core.services.users import ( 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, @@ -74,7 +54,7 @@ async def test_create_user_account_rejects_when_registration_disabled( "Email already exists.", ), ( - {"pubkey": _pubkey(1)}, + {"pubkey": f"{1:064x}"}, {"pubkey": lambda existing: existing.pubkey}, "Pubkey already exists.", ), @@ -397,3 +377,23 @@ async def test_check_register_activation_settings_handles_invitation_codes( "lnbits_register_one_time_activation_codes", previous_stored_codes.value if previous_stored_codes else original_one_time, ) + + +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, + )