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

refactor: extract helpers, move private functions to the bottom

Browse Source
This commit is contained in:
Vlad Stan
2026-03-25 10:49:52 +02:00
parent 2c84f9781d
commit 0fd116e7ee
15 changed files with 366 additions and 422 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/api/test_asset_api.py
+8 -32
View File
@@ -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)
tests/api/test_extension_api.py
+9 -48
View File
@@ -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,
tests/api/test_extensions_builder_api.py
+9 -50
View File
@@ -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",
tests/api/test_lnurl_api.py
+4 -15
View File
@@ -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")
)
tests/api/test_payment_api.py
+24 -24
View File
@@ -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
tests/api/test_wallet_api.py
+4 -4
View File
@@ -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"}
tests/helpers.py
+143 -1
View File
@@ -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()
tests/unit/test_services_assets.py
+13 -26
View File
@@ -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
tests/unit/test_services_extensions.py
+8 -35
View File
@@ -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
tests/unit/test_services_extensions_builder.py
+2 -34
View File
@@ -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",
tests/unit/test_services_funding_source.py
+8 -8
View File
@@ -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
tests/unit/test_services_lnurl.py
+21 -32
View File
@@ -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")
tests/unit/test_services_notifications.py
+47 -47
View File
@@ -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)
tests/unit/test_services_payments.py
+45 -45
View File
@@ -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
tests/unit/test_services_users.py
+21 -21
View File
@@ -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,
)